Writing NOC Commands

If you need to use a particular piece of code frequently, you can organize it into commands. These commands are executed by calling ./noc <command_name>.

For instance, to display the version, you can use the command ./noc about, which is located in the commands/about.py file.

[root@test noc]# ./noc about
22.2+noc-1968.161.9153a970

Command Structure

In general, the structure of a command looks like this:

from noc.core.management.base import BaseCommand
from noc.core.version import version


class Command(BaseCommand):
    def handle(self, *args, **options):
        self.print(version.version)


if __name__ == "__main__":
    Command().run()

Let's examine an example in detail.

from noc.core.management.base import BaseCommand
from noc.core.version import version


class Command(BaseCommand):
    def handle(self, *args, **options):
        self.print(version.version)


if __name__ == "__main__":
    Command().run()

Importing the base class for command.

from noc.core.management.base import BaseCommand
from noc.core.version import version


class Command(BaseCommand):
    def handle(self, *args, **options):
        self.print(version.version)


if __name__ == "__main__":
    Command().run()

Importing the structure, containing the NOC's version.

from noc.core.management.base import BaseCommand
from noc.core.version import version


class Command(BaseCommand):
    def handle(self, *args, **options):
        self.print(version.version)


if __name__ == "__main__":
    Command().run()

The implementation of our command should reside within a class derived from the base class BaseCommand.

from noc.core.management.base import BaseCommand
from noc.core.version import version


class Command(BaseCommand):
    def handle(self, *args, **options):
        self.print(version.version)


if __name__ == "__main__":
    Command().run()

The entry point for your command is within the handle function. Therefore, we must override this function to define our command's behavior.

from noc.core.management.base import BaseCommand
from noc.core.version import version


class Command(BaseCommand):
    def handle(self, *args, **options):
        self.print(version.version)


if __name__ == "__main__":
    Command().run()

The BaseCommand.print() function is used to print output to stdout when developing custom commands. It is recommended to always use this function instead of the built-in print() function within your commands.

In this case, we are using it to print the version information.

from noc.core.management.base import BaseCommand
from noc.core.version import version


class Command(BaseCommand):
    def handle(self, *args, **options):
        self.print(version.version)


if __name__ == "__main__":
    Command().run()

This part of the code is common to all commands and is responsible for launching our command from the command line.

Example with Argument Parsing

As an example, we will extract the code for checking configured metrics into a command. We will add the ability to pass a list of metrics as parameters. The code will be placed in the commands/check-metrics.py file.

# Python modules
import argparse

# NOC modules
from noc.core.management.base import BaseCommand
from noc.core.mongo.connection import connect
from noc.pm.models.metrictype import MetricType
from noc.sa.models.managedobjectprofile import ManagedObjectProfile
from noc.inv.models.interfaceprofile import InterfaceProfile


class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument("metrics", nargs=argparse.REMAINDER, help="Crashinfo UUIDs")

    def handle(self, *args, **options):
        if not options.get("metrics"):
            self.die("No requested metrics")
        connect()
        iface_metrics, object_metrics = [], []
        for m in options["metrics"]:
            mt = MetricType.get_by_name(m)
            if not mt:
                self.print(f"Unknown metric name '{m}'. Skipping")
                continue
            if mt.scope.name == "Interface":
                iface_metrics.append(mt)
                continue
            object_metrics.append(mt)
        if object_metrics:
            self.print("Checking Object Metric")
            self.check_object_metrics(object_metrics)
        if iface_metrics:
            self.print("Checking Interface Metric")
            self.check_interface_metrics(iface_metrics)

    def check_object_metrics(self, metrics: list[MetricType]):
        mt_check: dict[str, MetricType] = {str(mt.id): mt for mt in metrics}
        for mop in ManagedObjectProfile.objects.filter(
            enable_periodic_discovery_metrics=True, enable_periodic_discovery=True
        ):
            checks = set(mt_check)
            for mc in mop.metrics:
                if mc["metric_type"] in checks:
                    checks.remove(mc["metric_type"])
            if checks:
                self.print(
                    f"[{mop.name}] Not configured metrics: ",
                    ",".join(mt_check[c].name for c in checks),
                )

    def check_interface_metrics(self, metrics: list[MetricType]):
        for ip in InterfaceProfile.objects.filter(metrics__exists=True):
            checks = set(metrics)
            for mc in ip.metrics:
                if mc.metric_type in checks:
                    checks.remove(mc.metric_type)
            if checks:
                self.print(
                    f"[{ip.name}] Not configured metrics: ", ",".join(c.name for c in checks)
                )


if __name__ == "__main__":
    Command().run()

Let's run the following:

[root@test noc]# ./noc check-metrics 'Interface | Errors | In'
Checking Interface Metric
Not configured metrics on profile Аплинк:  Interface | Errors | In
Not configured metrics on profile Порт РРЛ:  Interface | Errors | In

Let's take a closer look at an example:

# Python modules
import argparse

# NOC modules
from noc.core.management.base import BaseCommand
from noc.core.mongo.connection import connect
from noc.pm.models.metrictype import MetricType
from noc.sa.models.managedobjectprofile import ManagedObjectProfile
from noc.inv.models.interfaceprofile import InterfaceProfile


class Command(BaseCommand):
    def add_arguments(self, parser):

Let's import the standard Python modules we'll need later.

# Python modules
import argparse

# NOC modules
from noc.core.management.base import BaseCommand
from noc.core.mongo.connection import connect
from noc.pm.models.metrictype import MetricType
from noc.sa.models.managedobjectprofile import ManagedObjectProfile
from noc.inv.models.interfaceprofile import InterfaceProfile


class Command(BaseCommand):
    def add_arguments(self, parser):

We also need to import some NOC definitions.

    def add_arguments(self, parser):
        parser.add_argument("metrics", nargs=argparse.REMAINDER, help="Crashinfo UUIDs")

    def handle(self, *args, **options):
        if not options.get("metrics"):

The implementation of our command should be within a class called Command, derived from the BaseCommand base class.

    def add_arguments(self, parser):
        parser.add_argument("metrics", nargs=argparse.REMAINDER, help="Crashinfo UUIDs")

    def handle(self, *args, **options):
        if not options.get("metrics"):

The add_arguments function allows us to configure the `argsparse`` command parser and set up parsing of additional arguments.

    def add_arguments(self, parser):
        parser.add_argument("metrics", nargs=argparse.REMAINDER, help="Crashinfo UUIDs")

    def handle(self, *args, **options):
        if not options.get("metrics"):

We configure the argument parser to place the entire command line remainder (REMAINDER) into the metrics option.

        if not options.get("metrics"):
            self.die("No requested metrics")
        connect()
        iface_metrics, object_metrics = [], []
        for m in options["metrics"]:
            mt = MetricType.get_by_name(m)
            if not mt:
                self.print(f"Unknown metric name '{m}'. Skipping")
                continue
            if mt.scope.name == "Interface":
                iface_metrics.append(mt)
                continue
            object_metrics.append(mt)
        if object_metrics:
            self.print("Checking Object Metric")
            self.check_object_metrics(object_metrics)
        if iface_metrics:
            self.print("Checking Interface Metric")
            self.check_interface_metrics(iface_metrics)

The entry point for the command is in the handle function, so we must override it.

        if not options.get("metrics"):
            self.die("No requested metrics")
        connect()
        iface_metrics, object_metrics = [], []
        for m in options["metrics"]:
            mt = MetricType.get_by_name(m)
            if not mt:
                self.print(f"Unknown metric name '{m}'. Skipping")
                continue
            if mt.scope.name == "Interface":
                iface_metrics.append(mt)
                continue
            object_metrics.append(mt)
        if object_metrics:
            self.print("Checking Object Metric")
            self.check_object_metrics(object_metrics)
        if iface_metrics:
            self.print("Checking Interface Metric")
            self.check_interface_metrics(iface_metrics)

We check if command line parameters are provided ( option["metrics"] ), and if not, we call the BaseCommand.die() function. The die() function prints an error message to stderr and terminates the command with a system error code.

        if not options.get("metrics"):
            self.die("No requested metrics")
        connect()
        iface_metrics, object_metrics = [], []
        for m in options["metrics"]:
            mt = MetricType.get_by_name(m)
            if not mt:
                self.print(f"Unknown metric name '{m}'. Skipping")
                continue
            if mt.scope.name == "Interface":
                iface_metrics.append(mt)
                continue
            object_metrics.append(mt)
        if object_metrics:
            self.print("Checking Object Metric")
            self.check_object_metrics(object_metrics)
        if iface_metrics:
            self.print("Checking Interface Metric")
            self.check_interface_metrics(iface_metrics)

The connect() function establishes a connection to the MongoDB database. It should be executed before any database operations.

        if not options.get("metrics"):
            self.die("No requested metrics")
        connect()
        iface_metrics, object_metrics = [], []
        for m in options["metrics"]:
            mt = MetricType.get_by_name(m)
            if not mt:
                self.print(f"Unknown metric name '{m}'. Skipping")
                continue
            if mt.scope.name == "Interface":
                iface_metrics.append(mt)
                continue
            object_metrics.append(mt)
        if object_metrics:
            self.print("Checking Object Metric")
            self.check_object_metrics(object_metrics)
        if iface_metrics:
            self.print("Checking Interface Metric")
            self.check_interface_metrics(iface_metrics)

We create two empty lists:

• Interface Metrics: interface_metrics
• Object Metrics: object_metrics

        if not options.get("metrics"):
            self.die("No requested metrics")
        connect()
        iface_metrics, object_metrics = [], []
        for m in options["metrics"]:
            mt = MetricType.get_by_name(m)
            if not mt:
                self.print(f"Unknown metric name '{m}'. Skipping")
                continue
            if mt.scope.name == "Interface":
                iface_metrics.append(mt)
                continue
            object_metrics.append(mt)
        if object_metrics:
            self.print("Checking Object Metric")
            self.check_object_metrics(object_metrics)
        if iface_metrics:
            self.print("Checking Interface Metric")
            self.check_interface_metrics(iface_metrics)

We iterate through the command line parameters ( option["metrics"] ), as there can be more than one.

        if not options.get("metrics"):
            self.die("No requested metrics")
        connect()
        iface_metrics, object_metrics = [], []
        for m in options["metrics"]:
            mt = MetricType.get_by_name(m)
            if not mt:
                self.print(f"Unknown metric name '{m}'. Skipping")
                continue
            if mt.scope.name == "Interface":
                iface_metrics.append(mt)
                continue
            object_metrics.append(mt)
        if object_metrics:
            self.print("Checking Object Metric")
            self.check_object_metrics(object_metrics)
        if iface_metrics:
            self.print("Checking Interface Metric")
            self.check_interface_metrics(iface_metrics)

We need to retrieve the object (database record) by its name. In NOC, to fetch a record by name, model methods like .get_by_name() are used. In addition to simplifying the code, .get_by_name() also provides caching, which can significantly enhance performance.

        if not options.get("metrics"):
            self.die("No requested metrics")
        connect()
        iface_metrics, object_metrics = [], []
        for m in options["metrics"]:
            mt = MetricType.get_by_name(m)
            if not mt:
                self.print(f"Unknown metric name '{m}'. Skipping")
                continue
            if mt.scope.name == "Interface":
                iface_metrics.append(mt)
                continue
            object_metrics.append(mt)
        if object_metrics:
            self.print("Checking Object Metric")
            self.check_object_metrics(object_metrics)
        if iface_metrics:
            self.print("Checking Interface Metric")
            self.check_interface_metrics(iface_metrics)

If the record is not found, .get_by_name() returns None. We use the check for None to ensure that the user has provided the correct metric name. If the user has provided an incorrect name, we print an error message and move on to processing the next metric.

        if not options.get("metrics"):
            self.die("No requested metrics")
        connect()
        iface_metrics, object_metrics = [], []
        for m in options["metrics"]:
            mt = MetricType.get_by_name(m)
            if not mt:
                self.print(f"Unknown metric name '{m}'. Skipping")
                continue
            if mt.scope.name == "Interface":
                iface_metrics.append(mt)
                continue
            object_metrics.append(mt)
        if object_metrics:
            self.print("Checking Object Metric")
            self.check_object_metrics(object_metrics)
        if iface_metrics:
            self.print("Checking Interface Metric")
            self.check_interface_metrics(iface_metrics)

At this point, we check if the metric scope is Interface. If it is, we add the metric to the iface_metrics list; otherwise, we add it to object_metrics.

        if not options.get("metrics"):
            self.die("No requested metrics")
        connect()
        iface_metrics, object_metrics = [], []
        for m in options["metrics"]:
            mt = MetricType.get_by_name(m)
            if not mt:
                self.print(f"Unknown metric name '{m}'. Skipping")
                continue
            if mt.scope.name == "Interface":
                iface_metrics.append(mt)
                continue
            object_metrics.append(mt)
        if object_metrics:
            self.print("Checking Object Metric")
            self.check_object_metrics(object_metrics)
        if iface_metrics:
            self.print("Checking Interface Metric")
            self.check_interface_metrics(iface_metrics)

If the user has provided at least one object metric, we call the check_object_metrics function and pass the object_metrics list as a parameter.

        if not options.get("metrics"):
            self.die("No requested metrics")
        connect()
        iface_metrics, object_metrics = [], []
        for m in options["metrics"]:
            mt = MetricType.get_by_name(m)
            if not mt:
                self.print(f"Unknown metric name '{m}'. Skipping")
                continue
            if mt.scope.name == "Interface":
                iface_metrics.append(mt)
                continue
            object_metrics.append(mt)
        if object_metrics:
            self.print("Checking Object Metric")
            self.check_object_metrics(object_metrics)
        if iface_metrics:
            self.print("Checking Interface Metric")
            self.check_interface_metrics(iface_metrics)

If the user has provided at least one interface metric, we call the check_interface_metrics function and pass the interface_metrics list as a parameter.

        mt_check: dict[str, MetricType] = {str(mt.id): mt for mt in metrics}
        for mop in ManagedObjectProfile.objects.filter(
            enable_periodic_discovery_metrics=True, enable_periodic_discovery=True
        ):
            checks = set(mt_check)
            for mc in mop.metrics:
                if mc["metric_type"] in checks:
                    checks.remove(mc["metric_type"])
            if checks:
                self.print(
                    f"[{mop.name}] Not configured metrics: ",
                    ",".join(mt_check[c].name for c in checks),
                )

Now, let's define the check_object_metrics function, which takes a list of object metrics as input.

        mt_check: dict[str, MetricType] = {str(mt.id): mt for mt in metrics}
        for mop in ManagedObjectProfile.objects.filter(
            enable_periodic_discovery_metrics=True, enable_periodic_discovery=True
        ):
            checks = set(mt_check)
            for mc in mop.metrics:
                if mc["metric_type"] in checks:
                    checks.remove(mc["metric_type"])
            if checks:
                self.print(
                    f"[{mop.name}] Not configured metrics: ",
                    ",".join(mt_check[c].name for c in checks),
                )

We build a dictionary mt_check, where the metric ID is used as the key, and the metric object is stored as the value.

        mt_check: dict[str, MetricType] = {str(mt.id): mt for mt in metrics}
        for mop in ManagedObjectProfile.objects.filter(
            enable_periodic_discovery_metrics=True, enable_periodic_discovery=True
        ):
            checks = set(mt_check)
            for mc in mop.metrics:
                if mc["metric_type"] in checks:
                    checks.remove(mc["metric_type"])
            if checks:
                self.print(
                    f"[{mop.name}] Not configured metrics: ",
                    ",".join(mt_check[c].name for c in checks),
                )

We retrieve all object profiles where periodic discovery is enabled and metrics are defined.

        mt_check: dict[str, MetricType] = {str(mt.id): mt for mt in metrics}
        for mop in ManagedObjectProfile.objects.filter(
            enable_periodic_discovery_metrics=True, enable_periodic_discovery=True
        ):
            checks = set(mt_check)
            for mc in mop.metrics:
                if mc["metric_type"] in checks:
                    checks.remove(mc["metric_type"])
            if checks:
                self.print(
                    f"[{mop.name}] Not configured metrics: ",
                    ",".join(mt_check[c].name for c in checks),
                )

We create a set check based on the keys in mt_check. In the future, we will remove the discovered metrics from it.

        mt_check: dict[str, MetricType] = {str(mt.id): mt for mt in metrics}
        for mop in ManagedObjectProfile.objects.filter(
            enable_periodic_discovery_metrics=True, enable_periodic_discovery=True
        ):
            checks = set(mt_check)
            for mc in mop.metrics:
                if mc["metric_type"] in checks:
                    checks.remove(mc["metric_type"])
            if checks:
                self.print(
                    f"[{mop.name}] Not configured metrics: ",
                    ",".join(mt_check[c].name for c in checks),
                )

We iterate through all the metrics defined in the profile and, if they are present in our check, we remove them from the check set.

        mt_check: dict[str, MetricType] = {str(mt.id): mt for mt in metrics}
        for mop in ManagedObjectProfile.objects.filter(
            enable_periodic_discovery_metrics=True, enable_periodic_discovery=True
        ):
            checks = set(mt_check)
            for mc in mop.metrics:
                if mc["metric_type"] in checks:
                    checks.remove(mc["metric_type"])
            if checks:
                self.print(
                    f"[{mop.name}] Not configured metrics: ",
                    ",".join(mt_check[c].name for c in checks),
                )

If there are remaining metrics in our check set, we write a message stating that they are not configured for the profile.

        for ip in InterfaceProfile.objects.filter(metrics__exists=True):
            checks = set(metrics)
            for mc in ip.metrics:
                if mc.metric_type in checks:
                    checks.remove(mc.metric_type)
            if checks:
                self.print(
                    f"[{ip.name}] Not configured metrics: ", ",".join(c.name for c in checks)
                )

Now, let's define the check_interface_metrics function, which takes a list of interface metrics as input.

        for ip in InterfaceProfile.objects.filter(metrics__exists=True):
            checks = set(metrics)
            for mc in ip.metrics:
                if mc.metric_type in checks:
                    checks.remove(mc.metric_type)
            if checks:
                self.print(
                    f"[{ip.name}] Not configured metrics: ", ",".join(c.name for c in checks)
                )

Now, let's define the check_interface_metrics function, which takes a list of interface metrics as input.

        for ip in InterfaceProfile.objects.filter(metrics__exists=True):
            checks = set(metrics)
            for mc in ip.metrics:
                if mc.metric_type in checks:
                    checks.remove(mc.metric_type)
            if checks:
                self.print(
                    f"[{ip.name}] Not configured metrics: ", ",".join(c.name for c in checks)
                )

We create a set check from all the input function parameters.

        for ip in InterfaceProfile.objects.filter(metrics__exists=True):
            checks = set(metrics)
            for mc in ip.metrics:
                if mc.metric_type in checks:
                    checks.remove(mc.metric_type)
            if checks:
                self.print(
                    f"[{ip.name}] Not configured metrics: ", ",".join(c.name for c in checks)
                )

We iterate through the metrics defined in the profile, and if they are present in our check, we remove them from the check set.

        for ip in InterfaceProfile.objects.filter(metrics__exists=True):
            checks = set(metrics)
            for mc in ip.metrics:
                if mc.metric_type in checks:
                    checks.remove(mc.metric_type)
            if checks:
                self.print(
                    f"[{ip.name}] Not configured metrics: ", ",".join(c.name for c in checks)
                )

If there are remaining metrics in our check set, we write a message stating that they are not configured for the profile.

    Command().run()

This part of the code is common to all commands and is responsible for running our command from the command line.