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 and edges 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

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 npu.

The following code demonstrates adding a rank attribute to every host instance that has a component with the name of npu.

Add a rank to each host npu

import pytest
import conftest
from infragraph import *
from infragraph.closfabric import ClosFabric
from infragraph.infragraph_service import InfraGraphService


@pytest.mark.asyncio
async def test_rank_annotations():
    """Test adding a rank attribute to every npu 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="npu filter")
    filter.choice = QueryNodeFilter.ID_FILTER
    filter.id_filter.operator = QueryNodeId.REGEX
    filter.id_filter.value = r"host\.\d+\.npu\.\d+"
    npu_response = service.query_graph(npu_request)

    # annotate the graph
    annotate_request = AnnotateRequest()
    for idx, match in enumerate(npu_response.node_matches):
        annotate_request.nodes.add(name=match.id, attribute="rank", value=str(idx))
    service.annotate_graph(annotate_request)

    # 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(annotate_request.nodes)
    assert len(annotate_request.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.

Add an ipaddress to each host mgmt component

import pytest
import conftest
import ipaddress
from infragraph import *
from infragraph.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
    annotate_request = AnnotateRequest()
    for idx, match in enumerate(nic_response.node_matches):
        annotate_request.nodes.add(
            name=match.id,
            attribute="ipaddress",
            value=str(ipaddress.ip_address(idx)),
        )
    service.annotate_graph(annotate_request)

    # 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(annotate_request.nodes)
    assert len(annotate_request.nodes) == len(ipaddress_response.node_matches)


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