The `annotate_graph` API

Overview

Once the infrastructure or system of systems has been defined by using the set_graph API the base graph can be extended by using the annotate_graph API to add additional data using nodes, edges, and links as endpoints for the data.

The main objective of the annotate_graph API is to separate the infrastructure model from specific use-case models by allowing the graph to be extended with any type of data. This ensures that InfraGraph does not morph into an attempt to define every nuance present in a system of systems.

Any annotation efforts can always be proposed as model or service enhancements by submitting issues or pull requests to the InfraGraph repository.

Additional Data

Some examples of additional data are: - AI data such as: - ranks - communication groups - Configuration data such as: - network interface card settings - device addresses - device routing tables

Annotation Structure

The annotate_graph API accepts an Annotation object that targets nodes, edges, or links in the graph. Each target supports one or more key-value attributes:

annotation = Annotation()

# add a node annotation
node = annotation.nodes.add(name="host.0.xpu.0")
node.attributes.add(attribute="rank", value="0")

# add an edge annotation
edge = annotation.edges.add(ep1="host.0.xpu.0", ep2="switch.0")
edge.attributes.add(attribute="bandwidth", value="400G")

# add a link annotation
link = annotation.links.add(name="pcie")
link.attributes.add(attribute="version", value="5.0")

service.annotate_graph(annotation)

Node Name Slicing

Node names support a slicing operator that expands to the fully qualified dot-separated format, making it easy to target ranges of nodes without enumerating each one individually:

Slice notation	Expands to
`server[0]xpu[0]`	`server.0.xpu.0`
`server[0:2]`	`server.0`, `server.1`
`server[0:2]xpu[0:3]`	`server.0.xpu.0`, `server.0.xpu.1`, `server.0.xpu.2`, `server.1.xpu.0`, ...
`switch`	`switch` (unchanged)

The following code examples demonstrates how to use the query_graph API in conjunction with the annotate_graph API to extend the graph with additional user specific data.

Adding `rank` data

In the Getting Started example, the instances of the Server device were created with the name of host and each instance having a specific number of components with a name of xpu.

The following code demonstrates adding a rank attribute to every host instance that has a component with the name of xpu. Each matching node is added to the Annotation object and a rank attribute is attached via attributes.add.

Add a rank to each host xpu

import pytest
from infragraph import *
from infragraph.blueprints.fabrics.closfabric import ClosFabric
from infragraph.infragraph_service import InfraGraphService


@pytest.mark.asyncio
async def test_rank_annotations():
    """Test adding a rank attribute to every xpu node"""
    # create the graph
    service = InfraGraphService()
    service.set_graph(ClosFabric())

    # query the graph for host npus
    npu_request = QueryRequest()
    filter = npu_request.node_filters.add(name="xpu filter")
    filter.choice = QueryNodeFilter.ID_FILTER
    filter.id_filter.operator = QueryNodeId.REGEX
    filter.id_filter.value = r"host\.\d+\.xpu\.\d+"
    npu_response = service.query_graph(npu_request)

    annotation = Annotation()
    for idx, match in enumerate(npu_response.node_matches):
        annotation_node = annotation.nodes.add(
            name=match.id
        )
        annotation_node.attributes.add(attribute="rank", value=str(idx))
    service.annotate_graph(annotation.serialize())

    # query the graph for rank attributes
    rank_request = QueryRequest()
    filter = rank_request.node_filters.add(name="rank filter")
    filter.choice = QueryNodeFilter.ATTRIBUTE_FILTER
    filter.attribute_filter.name = "rank"
    filter.attribute_filter.operator = QueryNodeId.REGEX
    filter.attribute_filter.value = r"\d+"
    rank_response = service.query_graph(rank_request)

    # validation
    assert len(npu_response.node_matches) > 0
    assert len(npu_response.node_matches) == len(annotation.nodes)
    assert len(annotation.nodes) == len(rank_response.node_matches)


if __name__ == "__main__":
    pytest.main(["-s", __file__])

Adding `ipaddress` data

In the Getting Started example, the instances of the Server device were created with the name of host and each instance having a mgmt nic component.

The following code demonstrates adding an ipaddress attribute to the host instance mgmt nic. Each matching node is added to the Annotation object and an ipaddress attribute is attached via attributes.add.

Add an ipaddress to each host mgmt component

import pytest
import ipaddress
from infragraph import *
from infragraph.blueprints.fabrics.closfabric import ClosFabric
from infragraph.infragraph_service import InfraGraphService


@pytest.mark.asyncio
async def test_ipaddress_annotations():
    """Test adding an ipaddress attribute to every server nic node"""
    # create the graph
    service = InfraGraphService()
    service.set_graph(ClosFabric())

    # query the graph for host nics
    npu_request = QueryRequest()
    filter = npu_request.node_filters.add(name="mgmt nic filter")
    filter.choice = QueryNodeFilter.ATTRIBUTE_FILTER
    filter.attribute_filter.name = "type"
    filter.attribute_filter.operator = QueryNodeId.EQ
    filter.attribute_filter.value = "mgmt-nic"
    nic_response = service.query_graph(npu_request)
    print(nic_response.node_matches)

    # annotate the graph
    annotation = Annotation()
    for idx, match in enumerate(nic_response.node_matches):
        annotation_node = annotation.nodes.add(
            name=match.id
        )
        annotation_node.attributes.add(attribute="ipaddress", value=str(ipaddress.ip_address(idx)))
    service.annotate_graph(annotation)

    # query the graph for ipaddress attributes
    ipaddress_request = QueryRequest()
    filter = ipaddress_request.node_filters.add(name="ipaddress filter")
    filter.choice = QueryNodeFilter.ATTRIBUTE_FILTER
    filter.attribute_filter.name = "ipaddress"
    filter.attribute_filter.operator = QueryNodeId.REGEX
    filter.attribute_filter.value = r".*"
    ipaddress_response = service.query_graph(ipaddress_request)
    print(ipaddress_response.node_matches)

    # validation
    assert len(nic_response.node_matches) > 0
    assert len(nic_response.node_matches) == len(annotation.nodes)
    assert len(annotation.nodes) == len(ipaddress_response.node_matches)


if __name__ == "__main__":
    pytest.main(["-s", __file__])

The annotate_graph API