Написание собственных команд NOC
При необходимости постоянного использования кода его реализацию можно вынести в команду. Команды запускаются вызовом ./noc <command_name>
.
Пример команда для вывода версии - ./noc about
расположена в commands/about.py
[root@test noc]# ./noc about
22.2+noc-1968.161.9153a970
Структура команды
В общем виде структура команды выглядит так:
sample.py |
---|
| 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 |
---|
| 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 |
---|
| 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 |
---|
| 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 |
---|
| 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 |
---|
| 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 |
---|
| 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 |
---|
| 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 |
---|
| 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 |
---|
| 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 |
---|
| if __name__ == "__main__":
Command().run()
|
Эта часть кода является общей для всех команд и отвечает за запуск нашей команды из командной строки.