Allocation

In a service lifecycle, allocation is the lifecycle stage where identifiers are allocated for use by a specific service instance.

For example a customer orders a virtual wire between two ports on two routers. The customer specifies router, port and vlan for both the A and Z side of the wire. In the network, this virtual wire is implemented as a VXlan tunnel, tied to both endpoints. Each such tunnel requires a “VXLAN Network Identifier (VNI)” that uniquely identifies the tunnel. In the allocation phase, the orchestrator selects a VNI and ensures no other customer is assigned the same VNI.

Correct allocation is crucial for the correct functioning of automated services. However, when serving multiple customers at once or when mediating between multiple inventories, correct allocation can be challenging, due to concurrency and distribution effects.

LSM offers a framework to perform allocation correctly and efficiently. The remainder of this document will explain how.

Types of Allocation

We distinguish several types of allocation. The next sections will explain each type, from simplest to most advanced. After the basic explanation, a more in-depth explanation is give for the different types. When first learning about LSM allocation (or allocation in general), it is important to have a basic understanding of the different types, before diving into the details.

LSM internal allocation

The easiest form of allocation is when no external inventory is involved. A range of available identifiers is assigned to LSM to distribute as it sees fit. For example, VNI range 50000-70000 is reserved to this service and can be used by LSM freely. This requires no coordination with external systems and is supported out-of-the-box.

The VNI example, allocation would look like this

main.cf

import lsm
import lsm::fsm

entity VlanAssignment extends lsm::ServiceEntity:
    string name

    int? vlan_id
    lsm::attribute_modifier vlan_id__modifier="r"
end

implement VlanAssignment using parents, do_deploy

binding = lsm::ServiceEntityBinding(
    service_entity="__config__::VlanAssignment",
    lifecycle=lsm::fsm::simple,
    service_entity_name="vlan-assignment",
    allocation_spec="allocate_vlan",
)

for assignment in lsm::all(binding):
    VlanAssignment(
        instance_id=assignment["id"],
        entity_binding=binding,
        **assignment["attributes"]
    )
end

The main changes in the model are:

1. the attributes that have to be allocated are added to the service definition as r (read only) attributes.

2. the service binding refers to an allocation spec (defined in python code)

plugins/__init__.py

"""
    Inmanta LSM

    :copyright: 2020 Inmanta
    :contact: code@inmanta.com
    :license: Inmanta EULA
"""
import inmanta_plugins.lsm.allocation as lsm

lsm.AllocationSpec(
    "allocate_vlan",
    lsm.LSM_Allocator(
        attribute="vlan_id", strategy=lsm.AnyUniqueInt(lower=50000, upper=70000)
    ),
)

The allocation spec specifies how to allocate the attribute:

1. Use the pure LSM internal allocation mechanism for vlan_id

2. To select a new value, use the AnyUniqueInt strategy, which selects a random number in the specified range

Internally, this works by storing allocations in read-only attributes on the instance. The lsm::all function ensures that if a value is already in the attribute, that value is used Otherwise the allocator gets an appropriate, new value, that doesn’t collide with any value in any attribute-set of any other service instance.

In practice, this means that a value is allocated as long as it’s in the active, candidate or rollback attribute sets of any non-terminated service instance. When a service instance is terminated, or clears one of its attribute sets, all identifiers are automatically deallocated.

Important note when designing custom lifecycles: allocation only happens during validating, and the result of the allocation is always written to the candidate attributes.

External lookup

Often, values received via the NorthBound API are not directly usable. For example, a router can be identified in the API by its name, but what is required is its management IP. The management IP can be obtained based on the name, through lookup in an inventory.

While lookup is not strictly allocation, it is in many ways similar.

The basic mechanism for external lookup is similar to internal allocation: the resolved value is stored in a read-only parameter. This is done to ensure that LSM remains stable, even if the inventory is down or corrupted. This also implies that if the inventory wants to change the value (i.e. router management IP is suddenly changed), it should notify LSM. LSM will not by itself pick up inventory changes. This notification mechanism is currently not supported yet.

An example with router management IP looks like this:

main.cf

import lsm
import lsm::fsm

entity VirtualWire extends lsm::ServiceEntity:
    string router_a
    int port_a
    int vlan_a
    string router_z
    int port_z
    int vlan_z
    int? vni
    std::ipv4_address?  router_a_mgmt_ip
    std::ipv4_address?  router_z_mgmt_ip
    lsm::attribute_modifier vni__modifier="r"
    lsm::attribute_modifier router_a_mgmt_ip__modifier="r"
    lsm::attribute_modifier router_z_mgmt_ip__modifier="r"
    lsm::attribute_modifier router_a__modifier="rw+"
    lsm::attribute_modifier router_z__modifier="rw+"
end

implement VirtualWire using parents, do_deploy

for assignment in lsm::all(binding):
  VirtualWire(
      instance_id=assignment["id"],
      router_a = assignment["attributes"]["router_a"],
      port_a = assignment["attributes"]["port_a"],
      vlan_a = assignment["attributes"]["vlan_a"],
      router_z = assignment["attributes"]["router_z"],
      port_z = assignment["attributes"]["port_z"],
      vlan_z = assignment["attributes"]["vlan_z"],
      vni=assignment["attributes"]["vni"],
      router_a_mgmt_ip=assignment["attributes"]["router_a_mgmt_ip"],
      router_z_mgmt_ip=assignment["attributes"]["router_z_mgmt_ip"],
      entity_binding=binding,
  )
end

binding = lsm::ServiceEntityBinding(
    service_entity="__config__::VirtualWire",
    lifecycle=lsm::fsm::simple,
    service_entity_name="virtualwire",
    allocation_spec="allocate_for_virtualwire",

While the allocation implementation could look like the following

plugins/__init__.py

"""
    Inmanta LSM

    :copyright: 2020 Inmanta
    :contact: code@inmanta.com
    :license: Inmanta EULA
"""
import os
from typing import Any, Optional

import inmanta_plugins.lsm.allocation as lsm
import psycopg2
from inmanta_plugins.lsm.allocation import (
    AllocationContext,
    ExternalAttributeAllocator,
    T,
)
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT


class PGRouterResolver(ExternalAttributeAllocator[T]):
    def __init__(self, attribute: str, id_attribute: str) -> None:
        super().__init__(attribute, id_attribute)
        self.conn = None
        self.database = None

    def pre_allocate(self):
        """Connect to postgresql"""
        host = os.environ.get("db_host", "localhost")
        port = os.environ.get("db_port")
        user = os.environ.get("db_user")
        self.database = os.environ.get("db_name", "allocation_db")
        self.conn = psycopg2.connect(
            host=host, port=port, user=user, dbname=self.database
        )
        self.conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)

    def post_allocate(self) -> None:
        """Close connection"""
        self.conn.close()

    def needs_allocation(
        self, ctx: AllocationContext, instance: dict[str, Any]
    ) -> bool:
        attribute_not_yet_allocated = super().needs_allocation(ctx, instance)
        id_attribute_changed = self._id_attribute_changed(instance)
        return attribute_not_yet_allocated or id_attribute_changed

    def _id_attribute_changed(self, instance: dict[str, Any]) -> bool:
        if instance["candidate_attributes"] and instance["active_attributes"]:
            return instance["candidate_attributes"].get(self.id_attribute) != instance[
                "active_attributes"
            ].get(self.id_attribute)
        return False

    def _get_value_from_result(self, result: Optional[tuple[T]]) -> Optional[T]:
        if result and result[0]:
            return result[0]
        return None

    def allocate_for_attribute(self, id_attribute_value: Any) -> T:
        with self.conn.cursor() as cursor:
            cursor.execute(
                "SELECT mgmt_ip FROM routers WHERE name=%s", (id_attribute_value,)
            )
            result = cursor.fetchone()
            allocated_value = self._get_value_from_result(result)
            if allocated_value:
                return allocated_value
            raise Exception("No ip address found for %s", str(id_attribute_value))


lsm.AllocationSpec(
    "allocate_for_virtualwire",
    PGRouterResolver(id_attribute="router_a", attribute="router_a_mgmt_ip"),
    PGRouterResolver(id_attribute="router_z", attribute="router_z_mgmt_ip"),
    lsm.LSM_Allocator(
        attribute="vni", strategy=lsm.AnyUniqueInt(lower=50000, upper=70000)
    ),
)

External inventory owns allocation

When allocating is owned externally, synchronization between LSM and the external inventory is crucial. If either LSM or the inventory fails, this should not lead to inconsistencies. In other words, LSM doesn’t only have to maintain consistency between different service instances, but also between itself and the inventory.

The basic mechanism for external allocation is similar to external lookup. One important difference is that we also write our allocation to the inventory.

For example, consider that there is an external Postgres Database that contains the allocation table. In the model, this will look exactly the same as in the case of internal allocation, in the code, it will look as follows

plugins/__init__.py

"""
    Inmanta LSM

    :copyright: 2020 Inmanta
    :contact: code@inmanta.com
    :license: Inmanta EULA
"""
import os
from typing import Optional
from uuid import UUID

import inmanta_plugins.lsm.allocation as lsm
import psycopg2
from inmanta_plugins.lsm.allocation import ExternalServiceIdAllocator, T
from psycopg2.extensions import ISOLATION_LEVEL_SERIALIZABLE


class PGServiceIdAllocator(ExternalServiceIdAllocator[T]):
    def __init__(self, attribute: str) -> None:
        super().__init__(attribute)
        self.conn = None
        self.database = None

    def pre_allocate(self):
        """Connect to postgresql"""
        host = os.environ.get("db_host", "localhost")
        port = os.environ.get("db_port")
        user = os.environ.get("db_user")
        self.database = os.environ.get("db_name", "allocation_db")
        self.conn = psycopg2.connect(
            host=host, port=port, user=user, dbname=self.database
        )
        self.conn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)

    def post_allocate(self) -> None:
        """Close connection"""
        self.conn.close()

    def _get_value_from_result(self, result: Optional[tuple[T]]) -> Optional[T]:
        if result and result[0]:
            return result[0]
        return None

    def allocate_for_id(self, serviceid: UUID) -> T:
        """Allocate in transaction"""
        with self.conn.cursor() as cursor:
            cursor.execute(
                "SELECT allocated_value FROM allocation WHERE attribute=%s AND owner=%s",
                (self.attribute, serviceid),
            )
            result = cursor.fetchone()
            allocated_value = self._get_value_from_result(result)
            if allocated_value:
                return allocated_value
            cursor.execute(
                "SELECT max(allocated_value) FROM allocation where attribute=%s",
                (self.attribute,),
            )
            result = cursor.fetchone()
            current_max_value = self._get_value_from_result(result)
            allocated_value = current_max_value + 1 if current_max_value else 1
            cursor.execute(
                "INSERT INTO allocation (attribute, owner, allocated_value) VALUES (%s, %s, %s)",
                (self.attribute, serviceid, allocated_value),
            )
            self.conn.commit()
            return allocated_value


lsm.AllocationSpec(
    "allocate_vlan",
    PGServiceIdAllocator(
        attribute="vlan_id",
    ),
)

What is important to notice is that the code first tries to see if an allocation has already happened. This is important in case there was a failure before LSM could commit the allocation. In general, LSM must be able to identify what has been allocated to it, in order to recover aborted operations. This is done either by attaching an identifier when performing allocation by knowing where the value will be stored in the inventory up front (e.g. the inventory contains a service model as well, LSM can find the VNI for a service by requesting the VNI for that service directly).

In the above example, the identifier is the same as the service instance id that LSM uses internally to identify an instance. An attribute of the instance can also be used to identify it in the external inventory, as the name attribute in the the example below.

plugins/__init__.py

"""
    Inmanta LSM

    :copyright: 2020 Inmanta
    :contact: code@inmanta.com
    :license: Inmanta EULA
"""
import os
from typing import Any, Optional

import inmanta_plugins.lsm.allocation as lsm
import psycopg2
from inmanta_plugins.lsm.allocation import ExternalAttributeAllocator, T
from psycopg2.extensions import ISOLATION_LEVEL_SERIALIZABLE


class PGAttributeAllocator(ExternalAttributeAllocator[T]):
    def __init__(self, attribute: str, id_attribute: str) -> None:
        super().__init__(attribute, id_attribute)
        self.conn = None
        self.database = None

    def pre_allocate(self):
        """Connect to postgresql"""
        host = os.environ.get("db_host", "localhost")
        port = os.environ.get("db_port")
        user = os.environ.get("db_user")
        self.database = os.environ.get("db_name", "allocation_db")
        self.conn = psycopg2.connect(
            host=host, port=port, user=user, dbname=self.database
        )
        self.conn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)

    def post_allocate(self) -> None:
        """Close connection"""
        self.conn.close()

    def _get_value_from_result(self, result: Optional[tuple[T]]) -> Optional[T]:
        if result and result[0]:
            return result[0]
        return None

    def allocate_for_attribute(self, id_attribute_value: Any) -> T:
        """Allocate in transaction"""
        with self.conn.cursor() as cursor:
            cursor.execute(
                "SELECT allocated_value FROM allocation WHERE attribute=%s AND owner=%s",
                (self.attribute, id_attribute_value),
            )
            result = cursor.fetchone()
            allocated_value = self._get_value_from_result(result)
            if allocated_value:
                return allocated_value
            cursor.execute(
                "SELECT max(allocated_value) FROM allocation where attribute=%s",
                (self.attribute,),
            )
            result = cursor.fetchone()
            current_max_value = self._get_value_from_result(result)
            allocated_value = current_max_value + 1 if current_max_value else 1
            cursor.execute(
                "INSERT INTO allocation (attribute, owner, allocated_value) VALUES (%s, %s, %s)",
                (self.attribute, id_attribute_value, allocated_value),
            )
            self.conn.commit()
            return allocated_value


lsm.AllocationSpec(
    "allocate_vlan",
    PGAttributeAllocator(attribute="vlan_id", id_attribute="name"),
)

Second, it is required that the inventory has a procedure to safely obtain ownership of an identifier. There must be some way LSM can definitely determine if it has correctly obtained an identifier. In the example, the database transaction ensures this. Many other mechanisms exist, but the inventory has to support at least one. Examples of possible transaction coordination mechanism are:

1. an API endpoint that atomically and consistently performs allocation,

2. database transaction

3. Compare-and-set style API (when updating a value, the old value is also passed along, ensuring no concurrent updates are possible)

4. API with version argument (like the LSM API itself, when updating a value, the version prior to update has to be passed along, preventing concurrent updates)

5. Locks and/or Leases (a value or part of the inventory can be locked or leased(locked for some time) prior to allocation, the lock ensures no concurrent modifications)

This scenario performs no de-allocation.

External inventory with deallocation

To ensure de-allocation on an external inventory is properly executed, it is not executed during compilation, but by a handler. This ensures that de-allocation is retried until it completes successfully.

The example below shows how allocation and de-allocation of a VLAN ID can be done using an external inventory. The handler of the PGAllocation entity performs the de-allocation. An instance of this entity is only constructed when the service instance is in the deallocating state.

vlan_assignment/model/_init.cf

import lsm
import lsm::fsm

entity VlanAssignment extends lsm::ServiceEntity:
    string name

    int? vlan_id
    lsm::attribute_modifier vlan_id__modifier="r"
end

implement VlanAssignment using parents, do_deploy
implement VlanAssignment using de_allocation when lsm::has_current_state(self, "deallocating")

entity PGAllocation extends std::PurgeableResource:
    """
        This entity ensures that an identifier allocated in PostgreSQL
        gets de-allocated when the service instance is removed.
    """
   string attribute
   std::uuid service_id
   string agent
end

implement PGAllocation using std::none

implementation de_allocation for VlanAssignment:
    """
        De-allocate the vlan_id identifier.
    """
    self.resources += PGAllocation(
        attribute="vlan_id",
        service_id=instance_id,
        purged=true,
        send_event=true,
        agent="internal",
        requires=self.requires,
        provides=self.provides,
    )
end

binding = lsm::ServiceEntityBinding(
    service_entity="vlan_assignment::VlanAssignment",
    lifecycle=lsm::fsm::simple_with_deallocation,
    service_entity_name="vlan-assignment",
    allocation_spec="allocate_vlan",
)

for assignment in lsm::all(binding):
    VlanAssignment(
        instance_id=assignment["id"],
        entity_binding=binding,
        **assignment["attributes"],
    )
end

The handler associated with the PGAllocation handler is shown in the code snippet below. Note that the handler doesn’t have an implementation for the create_resource() and the update_resource() method since they can never be called. The only possible operation is a delete operation.

vlan_assignment/plugins/__init__.py

"""
    Inmanta LSM

    :copyright: 2020 Inmanta
    :contact: code@inmanta.com
    :license: Inmanta EULA
"""
import os
from typing import Optional
from uuid import UUID

import psycopg2
from inmanta.agent import handler
from inmanta.agent.handler import CRUDHandlerGeneric as CRUDHandler
from inmanta.agent.handler import ResourcePurged, provider
from inmanta.resources import PurgeableResource, resource
from inmanta_plugins.lsm.allocation import AllocationSpec, ExternalServiceIdAllocator
from psycopg2.extensions import ISOLATION_LEVEL_SERIALIZABLE


class PGServiceIdAllocator(ExternalServiceIdAllocator[int]):
    def __init__(self, attribute: str) -> None:
        super().__init__(attribute)
        self.conn = None
        self.database = None

    def pre_allocate(self) -> None:
        """Connect to postgresql"""
        host = os.environ.get("db_host", "localhost")
        port = os.environ.get("db_port")
        user = os.environ.get("db_user")
        self.database = os.environ.get("db_name", "allocation_db")
        self.conn = psycopg2.connect(
            host=host, port=port, user=user, dbname=self.database
        )
        self.conn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)

    def post_allocate(self) -> None:
        """Close connection"""
        self.conn.close()

    def _get_value_from_result(self, result: Optional[tuple[int]]) -> Optional[int]:
        if result and result[0]:
            return result[0]
        return None

    def allocate_for_id(self, serviceid: UUID) -> int:
        """Allocate in transaction"""
        with self.conn.cursor() as cursor:
            cursor.execute(
                "SELECT allocated_value FROM allocation WHERE attribute=%s AND owner=%s",
                (self.attribute, serviceid),
            )
            result = cursor.fetchone()
            allocated_value = self._get_value_from_result(result)
            if allocated_value:
                return allocated_value
            cursor.execute(
                "SELECT max(allocated_value) FROM allocation where attribute=%s",
                (self.attribute,),
            )
            result = cursor.fetchone()
            current_max_value = self._get_value_from_result(result)
            allocated_value = current_max_value + 1 if current_max_value else 1
            cursor.execute(
                "INSERT INTO allocation (attribute, owner, allocated_value) VALUES (%s, %s, %s)",
                (self.attribute, serviceid, allocated_value),
            )
            self.conn.commit()
            return allocated_value

    def has_allocation_in_inventory(self, serviceid: UUID) -> bool:
        """
        Check whether a VLAN ID is allocated by the service instance with the given id.
        """
        with self.conn.cursor() as cursor:
            cursor.execute(
                "SELECT allocated_value FROM allocation WHERE attribute=%s AND owner=%s",
                (self.attribute, serviceid),
            )
            result = cursor.fetchone()
            allocated_value = self._get_value_from_result(result)
            if allocated_value:
                return True
            return False

    def de_allocate(self, serviceid: UUID) -> None:
        """
        De-allocate the VLAN ID allocated by the service instance with the given id.
        """
        with self.conn.cursor() as cursor:
            cursor.execute(
                "DELETE FROM allocation WHERE attribute=%s AND owner=%s",
                (self.attribute, serviceid),
            )
            self.conn.commit()


@resource("vlan_assignment::PGAllocation", agent="agent", id_attribute="service_id")
class PGAllocationResource(PurgeableResource):
    fields = ("attribute", "service_id")


@provider("vlan_assignment::PGAllocation", name="pgallocation")
class PGAllocation(CRUDHandler[PGAllocationResource]):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._allocator = PGServiceIdAllocator(attribute="vlan_id")

    def pre(self, ctx: handler.HandlerContext, resource: PGAllocationResource) -> None:
        self._allocator.pre_allocate()

    def post(self, ctx: handler.HandlerContext, resource: PGAllocationResource) -> None:
        self._allocator.post_allocate()

    def read_resource(
        self, ctx: handler.HandlerContext, resource: PGAllocationResource
    ) -> None:
        if not self._allocator.has_allocation_in_inventory(resource.service_id):
            raise ResourcePurged()

    def delete_resource(
        self, ctx: handler.HandlerContext, resource: PGAllocationResource
    ) -> None:
        self._allocator.de_allocate(resource.service_id)


AllocationSpec("allocate_vlan", PGServiceIdAllocator(attribute="vlan_id"))