Перейти к содержанию

Написание собственных команд NOC

При необходимости постоянного использования кода его реализацию можно вынести в команду. Команды запускаются вызовом ./noc <command_name>.

Пример команда для вывода версии - ./noc about расположена в commands/about.py

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

Структура команды

В общем виде структура команды выглядит так:

sample.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
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()

Рассмотрим пример детальнее:

sample.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
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()

Импортируем базовый класс команды.

sample.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
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()

Импортируем структуру, содержащую версию NOC

sample.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
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()

Реализалия нашей команды должна находиться в классе Command, порожденным от базового класса BaseCommand

sample.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
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()

Точка входа в команду находится в функции handle, поэтому мы должны ее переопределить.

sample.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
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()

Функция BaseCommand.print() печатает аргумент на stdout, всегда используйте ее, вместо встроенной функции print() в командах. В данном случае мы печатаем версию.

sample.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
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()

Эта часть кода является общей для всех команд и отвечает за запуск нашей команды из командной строки.

Пример c разбором аргументов

В качестве примера вынесем код проверки настроенных метрик в команду. Добавим возможность передавать список метрик через параметры. Расположим код в файле commands/check-metrics.py:

check-metrics.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
# Python modules
import argparse
from typing import List, Dict

# 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()

Запустим на исполнение:

[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

Рассмотрим пример детальнее:

check-metrics.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
# Python modules
import argparse
from typing import List, Dict

# 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):

Импортируем стандартные модули python, которые понадобятся нам в дальнейшем.

check-metrics.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
# Python modules
import argparse
from typing import List, Dict

# 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):

Нам также понадобятся импортировать несколько опеределений NOC.

check-metrics.py
13
14
15
16
17
class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument("metrics", nargs=argparse.REMAINDER, help="Crashinfo UUIDs")

    def handle(self, *args, **options):

Реализалия нашей команды должна находиться в классе Command, порожденным от базового класса BaseCommand

check-metrics.py
13
14
15
16
17
class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument("metrics", nargs=argparse.REMAINDER, help="Crashinfo UUIDs")

    def handle(self, *args, **options):

Функция add_arguments позволяет настроить парсер команд argsparse и настроить парсинг дополнительных аргументов.

check-metrics.py
13
14
15
16
17
class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument("metrics", nargs=argparse.REMAINDER, help="Crashinfo UUIDs")

    def handle(self, *args, **options):

Мы настраиваем парсер аргументов таким образом, чтобы весь остаток командной строки (REMAINDER) был помещен в опцию metrics

check-metrics.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
    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)

Точка входа в команду находится в функции handle, поэтому мы должны ее переопределить.

check-metrics.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
    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)
Мы проверяем, заданы ли парамеры командной строки (option["metrics"]), если нет - вызываем функцию BaseCommand.die(). Функция die() печатает сообщение об ошибке в stderr и завершает работу команды с системным кодом ошибки().

check-metrics.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
    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)
Функция connect() осуществляет соединение с базой mongodb. Она должна быть выполнена заранее до любого обращения к базе.

check-metrics.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
    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)
Мы создаем два пустых списка:
  • Метрики интерфейса: interface_metrics
  • Метрики объекта: object_metrics

check-metrics.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
    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)
Мы пробегаем по всем параметрам командной строки (option["metrics"]), так как их может быть больше одного.

check-metrics.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
    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)
Нам необходимо по имени метрики получить ее объект (запись в базе). В NOC для получения записи по имени используются методы моделей .get_by_name(). Помимо упрощения кода .get_by_name() также обеспечивает кеширование, что может сильно повышать производительность.

check-metrics.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
    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)
Если запись не найдена, .get_by_name() возвращает None. Мы используем проверку на None, чтобы убедиться, что пользователь правильно указал имя метрики. Если пользователь указал неверное имя, мы печатаем сообщение и переходим к обработке следующей метрики.

check-metrics.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
    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)
В этом месте мы проверяем, если metric scope - Interface, то добавляем метрику в список iface_metrics, в противном случае - в object_metrics.

check-metrics.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
    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)
Если пользователь задал хоть одну объектную метрику, вызываем функцию check_object_metrics и передаем ей в качестве параметра список object_metrics

check-metrics.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
    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)
Если пользователь задал хоть одну интерфейсную метрику, вызываем функцию check_interface_metrics и передаем ей в качестве параметра список interface_metrics

check-metrics.py
38
39
40
41
42
43
44
45
46
47
48
49
50
51
    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),
                )
Определим функцию check_object_metrics, которая принимает на вход список объектных метрик.
check-metrics.py
38
39
40
41
42
43
44
45
46
47
48
49
50
51
    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),
                )

Мы строим словарь mt_check, который в качестве ключа использует id метрики, а в качестве значения хранит объект метрики.

check-metrics.py
38
39
40
41
42
43
44
45
46
47
48
49
50
51
    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),
                )
Мы извлекаем все профили объекта, для которых включен periodic discovery и заданы метрики.

check-metrics.py
38
39
40
41
42
43
44
45
46
47
48
49
50
51
    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),
                )
Мы строим множество check по ключам mt_check. В дальнейшем мы будем убирать из него найденные метрики.

check-metrics.py
38
39
40
41
42
43
44
45
46
47
48
49
50
51
    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),
                )
Мы проходим по всем метрикам, заданным в профиле, и, если они присутствуют в нашем check, удаляем их из множества check.
check-metrics.py
38
39
40
41
42
43
44
45
46
47
48
49
50
51
    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),
                )

Если в нашем множестве checks остались мерики, пишем сообщение, что они не сконфигурированы для профиля.

check-metrics.py
53
54
55
56
57
58
59
60
61
62
    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)
                )
Определим функцию check_interface_metrics, которая принимает на вход список интерфейсных метрик.

check-metrics.py
54
55
56
57
58
59
60
61
62
63
    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)
                )
Мы проходим по всем профилям интерфейсов, для которых сконфигурированы метрики.

check-metrics.py
54
55
56
57
58
59
60
61
62
63
    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)
                )
Мы строим множество checks из всех элементов входных параметров функции
check-metrics.py
54
55
56
57
58
59
60
61
62
63
    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)
                )

Мы проходим по всем метрикам, заданным в профиле, и, если они присутствуют в нашем check, удаляем их из множества check.

check-metrics.py
54
55
56
57
58
59
60
61
62
63
    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)
                )

Если в нашем множестве checks остались мерики, пишем сообщение, что они не сконфигурированы для профиля.

check-metrics.py
65
66
if __name__ == "__main__":
    Command().run()

Эта часть кода является общей для всех команд и отвечает за запуск нашей команды из командной строки.