Creating a PDU handler

You can add your own PDU handler to DCImanager 6. To do this, create the handler code and upload it to the platform.

Preparing the environment

The PDU handler must be written in Python. We recommend using Python version 3.9.

You can create a handler based on an existing project. To copy a project, connect to the DCImanager 6 location server via SSH and run the command:

docker cp eservice_handler:/opt/ispsystem/equip_handler ./

Project directories can be useful when creating a handler:

/common — common auxiliary classes and functions.
/pdu_common — auxiliary classes and functions for working with the PDU;
/pdu_common/handlers — PDU handlers.

You can see the required Python libraries and their versions in the project requirements.txt file. To install the required libraries, run the command:

pip3 install -r requirements.txt

To check the data types in the project, we recommend using the mypy analyzer.

Creating the handler

Class for working with PDU

The PDU class is inherited from the class:

BaseSnmpPdu — for devices working via SNMP protocol;
BasePdu — for other devices.

The BaseSnmpPdu class contains methods of interacting with the PDU via SNMP:

snmp_get — execute a read request for a specific OID;
snmp_set — execute a modifying query for a certain OID;
snmp_walk — execute a request, the result of which is a list.

Example of class description

class ExampleHandlerSnmp(BaseSnmpPdu):
    """Example handler class."""

    def __init__(self, pdu_data: PduSnmpData):
        """__init__.

        Args:
            pdu_data (PduSnmpData): pdu connection data
        """
        super().__init__(pdu_data)

To get the data for connecting to the PDU, use the self.pdu_data method:

Example to get community

self.pdu_data.snmp_params[“community”]

Examples of SNMP requests to a PDU:

Getting the device name

result = self.snmp_get("1.3.6.1.2.1.1.5.0")
print(result.value)

Editing the device name

self.snmp_set("1.3.6.1.2.1.1.5.0", "test.pdu")

Obtaining device system data

for elem in self.snmp_walk("1.3.6.1.2.1.1"):
    print(elem.value)

Each handler file must contain the make_handler function. This function creates a handler object:

Example of function

def make_handler(pdu_data: pduSnmpData) -> BaseSnmpPdu:
    """Create pdu handler object.

    Args:
        pdu_data (PduSnmpData): pdu connection data
        for selected protocol.

    Returns:
        BasePdu: Initialized pdu handler object
    """
    return ExampleHandlerSnmp(pdu_data)

Methods for working with PDU

To allow DCImanager 6 to interact with the PDU, override the BasePdu class methods:

status — PDU polling;
port_up — enabling the port;
port_down — disabling the port;
statistic — gathering statistics.

For each method, there are argument types and return values that the platform expects. The equipment management service does not use raw json data when communicating with the PDU, but rather its object representation. For example, to enable a port using the port_up method, an object of the PduPortView class with the following properties is passed as a parameter:

identify — PDU port identifier;
power_status — PDU port status in DCImanager 6.

An object of the same class with the current port state in the power_status property is expected in the reply.

When overriding methods, specify the required format of queries and return values. All auxiliary views are described in the project file /pdu_common/helper.py .

Methods for working with port power consumption statistics

For the handler to work with PDU port power consumption statistics, you can use one of two variants of the PduPortStatisticView method:

current port power consumption in kWh:

PduPortStatisticView(
    port_identity=port.oid_index,
    port_consumption=port.value_float,
)

port power consumption in kWh, the value of which increases over time:

PduPortStatisticView(
    port_identity=port.oid_index,
    port_consumption_incr=port_energy.value_float,
)

You can specify only one of these methods. If you add both methods to the handler code, the handler will work with errors.

To obtain the total power consumption of all ports, use the methods:

for handlers with consumption:

pdu_statistic.total_consumption = pdu_statistic.sum_ports_consumption()

for handlers with consumption_incr:

pdu_statistic.total_consumption_incr = pdu_statistic.sum_ports_consumption()

Example of handler code

Example of code

from typing import Optional, Union

from common.logger import get_logger
from common.misc import waiter, InversionDict, Oid
from pdu_common.base_snmp_pdu import BaseSnmpPdu
from pdu_common.connection import PduSnmpData
from pdu_common.helper import PduView, PduPortView, EnumPowerStatus, PduStatisticView, PduPortStatisticView


OID_RPCM_ADMINISTRATIVE_STATE = Oid("1.3.6.1.4.1.46235.2.4.2.1.4")

# Amperage of each output port in mA
OID_RPCM_OUTPUT_MILLIAMPS = Oid("1.3.6.1.4.1.46235.2.4.2.1.11")

# Energy of each output port in kW/h
OID_RPCM_OUTPUT_ENERGY_KWH = Oid("1.3.6.1.4.1.46235.2.4.2.1.15")


PORT_POWER_DICT = InversionDict(
    {
        EnumPowerStatus.UP: 1,
        EnumPowerStatus.DOWN: 0,
    }
)


def make_handler(pdu_data: PduSnmpData) -> BaseSnmpPdu:
    """
    Returns APC PDU handler object
    :param pdu_data: Snmp PDU data object
    :return:
    """

    return Rpcm(pdu_data)


class Rpcm(BaseSnmpPdu):
    """PDU Rpcm work class"""

    def status(self) -> PduView:
        """
        GetPduPortsStatus
        :return: PduView
        """

        pdu_info = PduView()

        for port in self.snmp_walk(OID_RPCM_ADMINISTRATIVE_STATE):
            pdu_info.ports.append(
                PduPortView(
                    identity=port.oid_index,
                    power_status=PORT_POWER_DICT.get_by_value(port.value_int, EnumPowerStatus.UNKNOWN),
                )
            )

        return pdu_info

    def __get_port_status(
        self, pdu_port_data: PduPortView, wait_for_status: Optional[EnumPowerStatus] = None
    ) -> PduPortView:
        """
        Get current PDU port status
        :param pdu_port_data: PduPortView
        :return: Modified PduPortView
        """

        logging.info(f"Get status for PDU port '{pdu_port_data.identity}'")

        port_status_oid = OID_RPCM_ADMINISTRATIVE_STATE + pdu_port_data.identity

        if wait_for_status is not None:
            matcher = lambda response: response == wait_for_status
        else:
            matcher = lambda response: not isinstance(response, Exception)

        @waiter
        def get_status() -> Union[EnumPowerStatus, Exception]:
            """Waiter for PDU status"""
            try:
                res = self.snmp_get(oid=port_status_oid).value_int
                return PORT_POWER_DICT.get_by_value(res, EnumPowerStatus.UNKNOWN)
            except Exception as error:
                return error

        logging.info(f"Get status '{pdu_port_data.power_status}' for PDU port '{pdu_port_data.identity}'")

        # waiting for status change to expected
        pdu_port_data.power_status = get_status(matcher=matcher, timeout=10)
        return pdu_port_data

    def __port_change_status(self, pdu_port_data: PduPortView, power_status: EnumPowerStatus) -> PduPortView:
        """Changing PDU port power status.

        Args:
            pdu_port_data (PduPortView): port
            power_status (EnumPowerStatus): new power status

        Returns:
            PduPortView: final port status
        """

        self.snmp_set(
            oid=OID_RPCM_ADMINISTRATIVE_STATE + pdu_port_data.identity,
            value=PORT_POWER_DICT[power_status],
        )
        return self.__get_port_status(pdu_port_data, power_status)

    def port_up(self, pdu_port_data: PduPortView) -> PduPortView:
        """Port up"""
        return self.__port_change_status(pdu_port_data, EnumPowerStatus.UP)

    def port_down(self, pdu_port_data: PduPortView) -> PduPortView:
        """Port down"""
        return self.__port_change_status(pdu_port_data, EnumPowerStatus.DOWN)

    def statistic(self) -> PduStatisticView:
        """
        Get PDU statistic.
        :return: PduStatisticView
        """

        pdu_statistic = PduStatisticView()

        logging.info("Get PDU load in Amps...")
        pdu_statistic.load = sum(port.value_float for port in self.snmp_walk(OID_RPCM_OUTPUT_MILLIAMPS)) / 1000

        ports_consumption = self.snmp_walk(OID_RPCM_OUTPUT_ENERGY_KWH)

        logging.info("Get PDU total energy in kWh...")
        pdu_statistic.total_consumption = float(sum(port.value_float for port in ports_consumption))

        logging.info("Get PDU energy in kWh for every port...")
        for port in ports_consumption:
            pdu_statistic.ports.append(
                PduPortStatisticView(
                    port_identity=port.oid_index,
                    port_consumption=port.value_float,
                )
            )

        return pdu_statistic


logging = get_logger("rpcm")

Loading the handler into the platform

To load the handler into the platform:

Create a directory with the following structure:
```
handler_dir/
├── __init__.py
└── my_handler.py
```
Comments

handler_dir — directory name. There should not be a handler directory with the same name in the platform
__init__.py — the initialization file. If there is no such file, create an empty file with that name
my_handler.py — the handler file
Create a tar.gz archive with the following directory:
```
tar -czvf custom_handler.tar.gz handler_dir
```
Comments to the command

custom_handler.tar.gz — archive name
handler_dir –- the name of the created directory with the handler
Log in to DCImanager 6 with administrator permissions:
```
curl -o- -k https://example.com/api/auth/v4/public/token \
    -H "isp-box-instance: true" \
    -d '{
        "email": "<admin_email>",
        "password": "<admin_pass>"
    }'
```
Comments to the command

example.com — domain name of the servier with DCImanager 6
<admin_email> — email of the DCImanager 6 administrator
<admin_pass> — password of the DCImanager 6 administrator

A response message in the format below will be received:
```
{"id":"24","token":"24-cee181d2-ccaa-4b64-a229-234aa7a25db6"}
```
Save the value of the token parameter from the received response.
Create a description for the handler:
```
curl -o- -k https://example.com/api/eservice/v3/custom_equipment \
    -H "isp-box-instance: true" \
    -H "x-xsrf-token: <token>" \
    -d '{
        "device_type": "<device>",
        "handler": "<internal_handler_name>",
        "name": "<handler_name>",
        "protocol": ["<handler_protocol>"],
        "features": []
    }'
```
Comments to the command
example.com — domain name of the server with DCImanager 6
<token> — authorization token
<device> — device type. Possible values:
- switch;
- pdu;
- vpu
<internal_handler_name> — unique internal name of the handler
<handler_name> — the name of the handler to be displayed in the platform interface
<handler_protocol> — protocol for working with the handler. For example, snmp
features — parameters for displaying information in the user interface. You can pass an empty array to leave the information display unaffected, or you can add:
- port_aggregation — if the handler supports receiving information about aggregated ports;
- collect_mac - if the handler supports obtaining a list of MAC addresses of devices that are behind the switch ports
The response will contain the id of the created handler. Save this value.
```
{"id": 1}
```
Load the archive with the handler into the platform:
```
curl -o- -k https://example.com/api/eservice/v3/custom_equipment/<handler_id>/content \
    -H "isp-box-instance: true" \
    -H "x-xsrf-token: <token>" \
    -F "data=@custom_handler.tar.gz" \
    -F "handler_import=<import_path>"
```
Comments to the command

example.com — domain name of the server with DCImanager 6
<handler id> — id of the created handler
<token> — authorization token
custom_handler.tar.gz — name of the archive with the handler
<import_path> - relative path for import. For example, if the handler file my_handler.py is in the handler_dir directory, the relative path would be handler_dir.my_handler

You can also use this command to upload new handler versions to the platform.