可扩展遥测指南
可扩展遥测概述
设置开发环境
要访问遥测收集器（位于 aos_developer_sdk 存储库中），请联系 瞻博网络支持。将您开发的任何新收集器贡献到存储库中。
为了保持系统环境完整，我们建议您使用虚拟环境来隔离所需的 Python 包（用于开发和测试）。您可以从 https://support.juniper.net/support/downloads/?p=apstra/ 下载基本开发环境 aos_developer_sdk.run。要加载环境，请执行：
aos_developer_sdk$ bash aos_development_sdk.run
4d8bbfb90ba8: Loading layer [==================================================>] 217.6kB/217.6kB
7d54ea05a373: Loading layer [==================================================>] 4.096kB/4.096kB
e2e40f457231: Loading layer [==================================================>] 1.771MB/1.771MB
Loaded image: aos-developer-sdk:2.3.1-129
================================================================================
Loaded AOS Developer SDK Environment Container Image
aos-developer-sdk:2.3.1-129.
Container can be run by
docker run -it \
-v <path to aos developer_sdk cloned repo>:/aos_developer_sdk \
--name <container name> \
aos-developer-sdk:2.3.1-129
================================================================================
此命令加载 aos_developer_sdk Docker 映像。映像加载完成后，将打印用于启动环境的命令。按照命令指定的方式启动容器环境。要安装依赖项，请执行：
root@f2ece48bb2f1:/# cd /aos_developer_sdk/ root@f2ece48bb2f1:/aos_developer_sdk# make setup_env ...
现在，已为开发和测试收集器设置了环境。环境中还安装了 Apstra SDK 包，例如设备驱动程序和 REST 客户端。
开发收集器
若要开发遥测收集器，请 按顺序指定以下内容。
写收集器
Collector 是一个必须派生自 aos.sdk.system_agent.base_telemetry_collector 的类。BaseTelemetryCollector。使用逻辑重写收集器的 收集 方法，以便：
从设备收集数据
收集器中的设备驱动程序实例提供对设备执行命令的方法。例如，大多数 Apstra 设备驱动程序都提供用于执行命令和返回输出的方法get_jsonget_text。
aos_developer_sdk环境的设备驱动程序已预安装。您可以探索可用于收集数据的方法。例如：
>>> from aos.sdk.driver.eos import Device
>>> device = Device('172.20.180.10', 'admin', 'admin')
>>> device.open()
>>> pprint.pprint(device.get_json('show version'))
{u'architecture': u'i386',
u'bootupTimestamp': 1548302664.0,
u'hardwareRevision': u'',
u'internalBuildId': u'68f3ae78-65cb-4ed3-8675-0ff2219bf118',
u'internalVersion': u'4.20.10M-10040268.42010M',
u'isIntlVersion': False,
u'memFree': 3003648,
u'memTotal': 4011060,
u'modelName': u'vEOS',
u'serialNumber': u'',
u'systemMacAddress': u'52:54:00:ce:87:37',
u'uptime': 62620.55,
u'version': u'4.20.10M'}
>>> dir(device)
['AOS_VERSION_FILE', '__class__', '__delattr__', '__dict__', '__doc__',
'__format__', '__getattribute__', '__hash__', '__init__', '__module__',
'__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__',
'__sizeof__', '__str__', '__subclasshook__', '__weakref__', 'close',
'device_info', 'driver', 'execute', 'get_aos_server_ip',
'get_aos_version_related_info', 'get_device_aos_version',
'get_device_aos_version_number', 'get_device_info', 'get_json',
'get_text', 'ip_address', 'onbox', 'open', 'open_options', 'password',
'probe', 'set_device_info', 'upload_file', 'username']
解析数据
收集的数据需要根据 Apstra 框架和上面确定的服务模式进行解析和重新格式化。具有通用存储架构的收集器遵循以下结构：
{
"items": [
{
"identity": <key goes here>,
"value": <value goes here>,
},
{
"identity": <key goes here>,
"value": <value goes here>,
},
...
]
}
具有基于 IBA 的模式的收集器遵循以下结构：
[
{
"key": <key goes here>,
"value": <value goes here>,
},
{
"key": <key goes here>,
"value": <value goes here>,
},
...
]
在上述结构中，发布的数据有多个项目。每个项目都有一个键和一个值。例如，要发布接口特定信息，要发布到框架的每个接口都有一个标识/键值对。
如果要使用第三方包解析从设备获取的数据，请在路径中列出 Python 包和版本。
<aos_developer_sdk>/aosstdcollectors/requirements_<NOS>.txt.依赖项安装的软件包不会与 Apstra 软件使用的软件包冲突。Apstra 安装的软件包可在开发环境中找到 /etc/aos/python_dependency.txt 。
将数据发布到框架
根据所需的架构收集和分析数据时，将数据发布到框架。您可以使用 post_data 收集器中提供的方法。它接受一个参数，即应发布到框架的数据。
存储库中的文件夹 aos_developer_sdk/aosstdcollectors/aosstdcollectors 包含每个 NOS 的文件夹。将收集器添加到与 NOS 匹配的文件夹中。从 Apstra 版本 4.1.0 开始不再支持 Cumulus，但此示例仅供说明之用。例如，要为 Cumulus 编写收集器，请将收集器 aos_developer_sdk/aosstdcollectors/aosstdcollectors/cumulus添加到 ，并以服务名称命名文件。例如，如果服务名称为 interface_in_out_bytes，则将文件 interface_in_out_bytes.py命名为 。
除了定义收集器类之外，还要在收集器文件中定义函数 collector_plugin 。该函数接受一个参数并返回实现的收集器类。
例如，基于通用存储架构的收集器如下所示：
"""
Service Name: interface_in_out_bytes
Schema:
Key: String, represents interface name.
Value: Json String with two possible keys:
rx: integer value, represents received bytes.
tx: integer value, represents transmitted bytes.
DOS: eos
Data collected using command: 'show interfaces'
Type of Collector: BaseTelemetryCollector
Storage Schema Path: aos.sdk.telemetry.schemas.generic
Application Schema: {
'type': 'object',
'properties': {
'identity': {
'type': 'string',
},
'value': {
'type': 'object',
'properties': {
'rx': {
'type': 'number',
},
'tx': {
'type': 'number',
}
},
'required': ['rx', 'tx'],
}
}
}
"""
import json
from aos.sdk.system_agent.base_telemetry_collector import BaseTelemetryCollector
# Inheriting from BaseTelemetryCollector
class InterfaceRxTxCollector(BaseTelemetryCollector):
# Overriding collect method
def collect(self):
# Obtaining the command output using the device instance.
collected_data = self.device.get_json('show interfaces')
# Data is in the format
# "interfaces": {
# "<interface_name>": {
# ....
# "interfaceCounters": {
# ....
# "inOctets": int
# "outOctets": int
# ....
# }
# }
# ...
# }
# Parse the data as per the schema and structure required.
parsed_data = json.dumps({
'items': [
{
'identity': intf_name,
'value': json.dumps({
'rx': intf_stats['interfaceCounters'].get('inOctets'),
'tx': intf_stats['interfaceCounters'].get('outOctets'),
})
} for intf_name, intf_stats in collected_data['interfaces'].iteritems()
if 'interfaceCounters' in intf_stats
]
})
# Post the data to the framework
self.post_data(parsed_data)
# Define collector_plugin class to return the Collector
def collector_plugin(_device):
return InterfaceRxTxCollector
基于 IBA 存储模式的收集器如下所示：
"""
Service Name: iba_bgp
Schema:
Key: JSON String, specifies local IP and peer IP.
Value: String. ‘1’ if state is established ‘2’ otherwise
DOS: eos
Data collected using command: 'show ip bgp summary vrf all'
Storage Schema Path: aos.sdk.telemetry.schemas.iba_string_data
Application Schema: {
'type': 'object',
'properties': {
key: {
'type': 'object',
'properties': {
'local_ip': {
'type': 'string',
},
'peer_ip': {
'type': 'string',
}
},
'required': ['local_ip', 'peer_ip'],
},
'value': {
'type': 'string',
}
}
}
"""
from aos.sdk.system_agent.base_telemetry_collector import IBATelemetryCollector
def parse_text_output(collected):
result = [
{'key': {'local_ip': str(vrf_info['routerId']), 'peer_ip': str(peer_ip)},
'value': str(
1 if session_info['peerState'] == 'Established' else 2)}
for vrf_info in collected['vrfs'].itervalues()
for peer_ip, session_info in vrf_info['peers'].iteritems()]
return result
# Inheriting from BaseTelemetryCollector
class IbaBgpCollector(BaseTelemetryCollector):
# Overriding collect method
def collect(self):
# Obtaining the command output using the device instance.
collected_data = self.device.get_json('show ip bgp summary vrf all')
# Parse the data as per the schema and structure required and
# post to framework.
self.post_data(parse_text_output(collected_data))
# Define collector_plugin class to return the Collector
def collector_plugin(device):
return IbaBgpCollector
单元测试收集器
存储库中的文件夹 aos_developer_sdk/aosstdcollectors/test 包含基于 NOS 的文件夹。将测试添加到与 NOS 匹配的文件夹中。例如，对积云收集器的测试将添加到 aos_developer_sdk/aosstdcollectors/test/cumulus中。建议使用前缀 test_命名单元测试。
现有基础结构实现用于模拟设备驱动程序命令响应的 Pytest 夹具 collector_factory 。测试开发的一般流程如下。
- 使用收集器工厂获取收集器实例和模拟的 Apstra 框架。收集器工厂将您编写的收集器类作为输入。
- 模拟设备响应。
- 调用收集方法。
- 验证发布到模拟 Apstra 框架的数据。
例如，测试如下所示：
import json
from aosstdcollectors.eos.interface_in_out_bytes import InterfaceRxTxCollector
# Test method with prefix 'test_'
def test_sanity(collector_factory):
# Using collector factory to retrieve the collector instance and mocked
# Apstra framework.
collector, mock_framework = collector_factory(InterfaceRxTxCollector)
command_response = {
'interfaces': {
'Ethernet1': {
'interfaceCounters': {
'inOctets': 10,
'outOctets': 20,
}
},
'Ethernet2': {
'interfaceCounters': {
'inOctets': 30,
'outOctets': 40,
}
}
}
}
# Set the device get_json method to retrieve the command response.
collector.device.get_json.side_effect = lambda _: command_response
# Invoke the collect method
collector.collect()
expected_data = [
{
'identity': 'Ethernet1',
'value': json.dumps({
'rx': 10,
'tx': 20,
}),
},
{
'identity': 'Ethernet2',
'value': json.dumps({
'rx': 30,
'tx': 40,
})
}
]
# validate the data posted by the collector
data_posted_by_collector = json.loads(mock_framework.post_data.call_args[0][0])
assert sorted(expected_data) == sorted(data_posted_by_collector["items"])
要运行测试，请执行：
root@1df9bf89aeaf:/aos_developer_sdk# make test root@1df9bf89aeaf:/aos_developer_sdk# make test
此命令执行存储库中的所有测试。
包收集器
所有收集器均基于 NOS 进行打包。要生成所有包，请在 执行 make aos_develop_sdk。您可以在 上 aos_developer_sdk/dist找到构建包。包构建大致可分为：
| 包 | 说明 |
|---|---|
| 内置收集器包 | 这些包具有前缀 aosstdcollectors_builtin_。要根据参考设计从设备收集遥测数据，Apstra 需要本节 <deviceblah> 中列出的服务。内置收集器包包含这些服务的收集器。这些包是基于每个 NOS 生成的。 |
| 自定义收集器包 | 这些软件包的名称中带有前缀 aosstdcollectors_custom_ 。这些包是基于每个 NOS 生成的。名为 aosstdcollectors_custom_<NOS>-0.1.0-py2-none-any.whl 的软件包包含开发的收集器。 |
| Apstra SDK 设备驱动程序包 | 这些包具有前缀 apstra_devicedriver_。这些包是基于每个 NOS 生成的。系统会为 NOS 生成在 Apstra 中默认不可用的软件包。 |
上传包
如果 Apstra 软件未提供适用于您的设备操作系统 （NOS） 的内置收集器包和 Apstra SDK 设备驱动程序，则必须将它们上传到 Apstra 服务器。
如果您使用的是异机解决方案，并且您的 NOS 不是 EOS，则必须上传内置收集器包。
上传包含收集器的软件包，并将其分配给设备系统代理或系统代理配置文件。
使用遥测收集器
设置遥测服务注册表
注册表将服务映射到其应用程序架构和存储架构路径。可以使用 REST 终结点 /api/telemetry-service-registry管理遥测服务注册表。如果不为特定服务添加注册表项，则无法为服务启用收集器。在使用服务时，无法修改服务的注册表项。
执行 make时，所有应用程序模式都打包到 dist 文件夹中的 tar 文件 （json_schemas.tgz） 中。使用 apstra-cli，您可以选择导入 .tgz 文件中的所有架构。
启动收集器
要启动服务，请将 POST API /api/systems/<system_id>/services 与以下三个参数一起使用：
| 参数 | |
|---|---|
| Input_data | 作为输入提供给收集器的数据。默认值为“无”。 |
| 区间 | 运行服务的时间间隔。默认为 120 秒。 |
| 名字 | 服务的名称。 |
您还可以通过 apstra-cli 实用程序管理收集器。
删除收集器
要删除服务，请使用 删除 API /api/systems/<system_id>/services/<service_name>。
获取收集的数据
要检索收集的数据，请使用 GET API /api/systems/<system_id>/services/<service_name>/data。仅保存上次迭代中收集的数据。数据不会在 Apstra 重启后保留。
列出正在运行的收集器服务
要检索设备上启用的服务列表，请使用 GET API /api/systems/<system_id>/services。