| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package snmptopology |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | _ "embed" |
| 8 | "fmt" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/gosnmp/gosnmp" |
| 12 | |
| 13 | topologyengine "github.com/netdata/netdata/go/plugins/pkg/l2topology" |
| 14 | |
| 15 | "github.com/netdata/netdata/go/plugins/pkg/funcapi" |
| 16 | "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi" |
| 17 | "github.com/netdata/netdata/go/plugins/plugin/go.d/collector/snmp/ddsnmp" |
| 18 | "github.com/netdata/netdata/go/plugins/plugin/go.d/collector/snmp/ddsnmp/ddsnmpcollector" |
| 19 | "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/snmputils" |
| 20 | ) |
| 21 | |
| 22 | //go:embed "config_schema.json" |
| 23 | var configSchema string |
| 24 | |
| 25 | func init() { |
| 26 | collectorapi.Register("snmp_topology", collectorapi.Creator{ |
| 27 | JobConfigSchema: configSchema, |
| 28 | Defaults: collectorapi.Defaults{ |
| 29 | UpdateEvery: 60, |
| 30 | }, |
| 31 | Create: func() collectorapi.CollectorV1 { return New() }, |
| 32 | Config: func() any { return &Config{} }, |
| 33 | }) |
| 34 | |
| 35 | // Register the topology function handler and method config so the snmp module |
| 36 | // can serve topology:snmp requests under the snmp:topology:snmp function name. |
| 37 | ddsnmp.TopologyHandler = &funcTopology{} |
| 38 | cfg := topologyMethodConfig() |
| 39 | ddsnmp.TopologyMethodConfig = &cfg |
| 40 | } |
| 41 | |
| 42 | func New() *Collector { |
| 43 | return &Collector{ |
| 44 | deviceCaches: make(map[string]*topologyCache), |
| 45 | deviceLastCollected: make(map[string]time.Time), |
| 46 | newSnmpClient: gosnmp.NewHandler, |
| 47 | newDdSnmpColl: func(cfg ddsnmpcollector.Config) ddCollector { |
| 48 | return ddsnmpcollector.New(cfg) |
| 49 | }, |
| 50 | } |
| 51 | } |
| 52 | |
| 53 | type ( |
| 54 | Collector struct { |
| 55 | collectorapi.Base `yaml:",inline"` |
| 56 | Config `yaml:",inline"` |
| 57 | |
| 58 | charts *collectorapi.Charts |
| 59 | deviceCaches map[string]*topologyCache // one cache per SNMP device |
| 60 | deviceLastCollected map[string]time.Time // last collection time per device |
| 61 | topologyCache *topologyCache // current device cache (set during refreshDeviceTopology) |
| 62 | topologyChartsAdded bool |
| 63 | |
| 64 | newSnmpClient func() gosnmp.Handler |
| 65 | newDdSnmpColl func(ddsnmpcollector.Config) ddCollector |
| 66 | } |
| 67 | ddCollector interface { |
| 68 | Collect() ([]*ddsnmp.ProfileMetrics, error) |
| 69 | } |
| 70 | ) |
| 71 | |
| 72 | func (c *Collector) Configuration() any { |
| 73 | return c.Config |
| 74 | } |
| 75 | |
| 76 | func (c *Collector) Init(context.Context) error { |
| 77 | return nil |
| 78 | } |
| 79 | |
| 80 | func (c *Collector) Check(context.Context) error { |
| 81 | return nil |
| 82 | } |
| 83 | |
| 84 | func (c *Collector) Charts() *collectorapi.Charts { |
| 85 | if c.charts == nil { |
| 86 | c.charts = &collectorapi.Charts{} |
| 87 | } |
| 88 | return c.charts |
| 89 | } |
| 90 | |
| 91 | func (c *Collector) Collect(context.Context) map[string]int64 { |
| 92 | if devices := ddsnmp.DeviceRegistry.Devices(); len(devices) > 0 { |
| 93 | refreshEvery := c.refreshEvery() |
| 94 | now := time.Now() |
| 95 | seen := make(map[string]bool, len(devices)) |
| 96 | |
| 97 | for _, dev := range devices { |
| 98 | key := fmt.Sprintf("%s:%d", dev.Hostname, dev.Port) |
| 99 | seen[key] = true |
| 100 | |
| 101 | lastCollected, exists := c.deviceLastCollected[key] |
| 102 | isNew := !exists |
| 103 | isStale := exists && now.Sub(lastCollected) >= refreshEvery |
| 104 | |
| 105 | if isNew || isStale { |
| 106 | c.refreshDeviceTopology(key, dev) |
| 107 | c.deviceLastCollected[key] = now |
| 108 | } |
| 109 | } |
| 110 | |
| 111 | c.pruneStaleDeviceCaches(seen) |
| 112 | } |
| 113 | |
| 114 | mx := make(map[string]int64) |
| 115 | c.collectTopologyMetrics(mx) |
| 116 | return mx |
| 117 | } |
| 118 | |
| 119 | const defaultRefreshEvery = 30 * time.Minute |
| 120 | |
| 121 | func (c *Collector) refreshEvery() time.Duration { |
| 122 | if d := c.RefreshEvery.Duration(); d > 0 { |
| 123 | return d |
| 124 | } |
| 125 | return defaultRefreshEvery |
| 126 | } |
| 127 | |
| 128 | func (c *Collector) Cleanup(context.Context) { |
| 129 | for key, cache := range c.deviceCaches { |
| 130 | snmpTopologyRegistry.unregister(cache) |
| 131 | delete(c.deviceCaches, key) |
| 132 | } |
| 133 | } |
| 134 | |
| 135 | // refreshDeviceTopology collects topology data for a single device into its own cache. |
| 136 | func (c *Collector) refreshDeviceTopology(key string, dev ddsnmp.DeviceConnectionInfo) { |
| 137 | snmpClient, err := newSNMPClientFromDeviceInfo(c.newSnmpClient, dev) |
| 138 | if err != nil { |
| 139 | c.Warningf("device '%s': failed to create SNMP client: %v", dev.Hostname, err) |
| 140 | return |
| 141 | } |
| 142 | if dev.MaxRepetitions != 0 { |
| 143 | snmpClient.SetMaxRepetitions(dev.MaxRepetitions) |
| 144 | } |
| 145 | if err := snmpClient.Connect(); err != nil { |
| 146 | c.Warningf("device '%s': failed to connect: %v", dev.Hostname, err) |
| 147 | return |
| 148 | } |
| 149 | defer func() { _ = snmpClient.Close() }() |
| 150 | |
| 151 | profiles := c.findTopologyProfiles(dev) |
| 152 | if len(profiles) == 0 { |
| 153 | return |
| 154 | } |
| 155 | |
| 156 | coll := c.newDdSnmpColl(ddsnmpcollector.Config{ |
| 157 | SnmpClient: snmpClient, |
| 158 | Profiles: profiles, |
| 159 | Log: c.Logger, |
| 160 | SysObjectID: dev.SysObjectID, |
| 161 | DisableBulkWalk: dev.DisableBulkWalk, |
| 162 | }) |
| 163 | |
| 164 | pms, err := coll.Collect() |
| 165 | if err != nil { |
| 166 | c.Warningf("device '%s': topology collection failed: %v", dev.Hostname, err) |
| 167 | return |
| 168 | } |
| 169 | |
| 170 | sysUptime, err := snmputils.GetSysUptime(snmpClient) |
| 171 | if err != nil { |
| 172 | c.Debugf("device '%s': failed to query system uptime: %v", dev.Hostname, err) |
| 173 | } |
| 174 | |
| 175 | // Build the next snapshot off-registry. Function readers keep seeing the |
| 176 | // previous complete snapshot until this collection is fully ingested. |
| 177 | next := c.newDeviceCollectionCache(dev) |
| 178 | c.topologyCache = next |
| 179 | defer func() { c.topologyCache = nil }() |
| 180 | |
| 181 | c.updateTopologySysUptime(sysUptime) |
| 182 | c.updateTopologyProfileTags(pms) |
| 183 | c.ingestTopologyProfileMetrics(pms) |
| 184 | c.collectTopologyVTPVLANContexts(dev) |
| 185 | c.finalizeTopologyCache() |
| 186 | |
| 187 | cache := c.getOrCreateDeviceCache(key) |
| 188 | cache.mu.Lock() |
| 189 | cache.replaceWith(next) |
| 190 | cache.mu.Unlock() |
| 191 | } |
| 192 | |
| 193 | func (c *Collector) getOrCreateDeviceCache(key string) *topologyCache { |
| 194 | cache, ok := c.deviceCaches[key] |
| 195 | if !ok { |
| 196 | cache = newTopologyCache() |
| 197 | c.deviceCaches[key] = cache |
| 198 | snmpTopologyRegistry.register(cache) |
| 199 | } |
| 200 | return cache |
| 201 | } |
| 202 | |
| 203 | func (c *Collector) newDeviceCollectionCache(dev ddsnmp.DeviceConnectionInfo) *topologyCache { |
| 204 | cache := newTopologyCache() |
| 205 | cache.updateTime = time.Now() |
| 206 | cache.staleAfter = c.refreshEvery() + time.Duration(c.UpdateEvery*2)*time.Second |
| 207 | cache.agentID = dev.Hostname |
| 208 | cache.localDevice = buildLocalTopologyDevice(dev) |
| 209 | return cache |
| 210 | } |
| 211 | |
| 212 | func (c *Collector) pruneStaleDeviceCaches(seen map[string]bool) { |
| 213 | for key, cache := range c.deviceCaches { |
| 214 | if !seen[key] { |
| 215 | snmpTopologyRegistry.unregister(cache) |
| 216 | delete(c.deviceCaches, key) |
| 217 | delete(c.deviceLastCollected, key) |
| 218 | } |
| 219 | } |
| 220 | } |
| 221 | |
| 222 | func (c *Collector) findTopologyProfiles(dev ddsnmp.DeviceConnectionInfo) []*ddsnmp.Profile { |
| 223 | return ddsnmp.DefaultCatalog().Resolve(ddsnmp.ResolveRequest{ |
| 224 | SysObjectID: dev.SysObjectID, |
| 225 | SysDescr: dev.SysDescr, |
| 226 | ManualProfiles: dev.ManualProfiles, |
| 227 | ManualPolicy: ddsnmp.ManualProfileAugment, |
| 228 | }).Project(ddsnmp.ConsumerTopology).Profiles() |
| 229 | } |
| 230 | |
| 231 | func (c *Collector) ingestTopologyProfileMetrics(pms []*ddsnmp.ProfileMetrics) { |
| 232 | for _, pm := range pms { |
| 233 | c.ingestTopologyMetricSet(pm.TopologyMetrics) |
| 234 | } |
| 235 | } |
| 236 | |
| 237 | func (c *Collector) ingestTopologyMetricSet(metrics []ddsnmp.Metric) { |
| 238 | for _, metric := range metrics { |
| 239 | c.updateTopologyCacheEntry(metric) |
| 240 | } |
| 241 | } |
| 242 | |
| 243 | // collectTopologyMetrics reads the aggregated topology from the global registry. |
| 244 | func (c *Collector) collectTopologyMetrics(mx map[string]int64) { |
| 245 | if !c.topologyChartsAdded { |
| 246 | c.addTopologyCharts() |
| 247 | c.topologyChartsAdded = true |
| 248 | } |
| 249 | |
| 250 | data, ok := snmpTopologyRegistry.snapshot() |
| 251 | if !ok { |
| 252 | mx["snmp_topology_devices_total"] = 0 |
| 253 | mx["snmp_topology_devices_discovered"] = 0 |
| 254 | mx["snmp_topology_links_total"] = 0 |
| 255 | mx["snmp_topology_links_lldp"] = 0 |
| 256 | mx["snmp_topology_links_cdp"] = 0 |
| 257 | mx["snmp_topology_links_stp"] = 0 |
| 258 | return |
| 259 | } |
| 260 | |
| 261 | totalDevices := 0 |
| 262 | for _, actor := range data.Actors { |
| 263 | if topologyengine.IsDeviceActorType(actor.ActorType) { |
| 264 | totalDevices++ |
| 265 | } |
| 266 | } |
| 267 | |
| 268 | var lldpLinks, cdpLinks, stpLinks int64 |
| 269 | for _, link := range data.Links { |
| 270 | switch link.Protocol { |
| 271 | case "lldp": |
| 272 | lldpLinks++ |
| 273 | case "cdp": |
| 274 | cdpLinks++ |
| 275 | case "stp": |
| 276 | stpLinks++ |
| 277 | } |
| 278 | } |
| 279 | |
| 280 | mx["snmp_topology_devices_total"] = int64(totalDevices) |
| 281 | mx["snmp_topology_devices_discovered"] = int64(maxInt(totalDevices-1, 0)) |
| 282 | mx["snmp_topology_links_total"] = int64(len(data.Links)) |
| 283 | mx["snmp_topology_links_lldp"] = lldpLinks |
| 284 | mx["snmp_topology_links_cdp"] = cdpLinks |
| 285 | mx["snmp_topology_links_stp"] = stpLinks |
| 286 | } |
| 287 | |
| 288 | func newSNMPClientFromDeviceInfo(newClient func() gosnmp.Handler, dev ddsnmp.DeviceConnectionInfo) (gosnmp.Handler, error) { |
| 289 | client := newClient() |
| 290 | |
| 291 | client.SetTarget(dev.Hostname) |
| 292 | client.SetPort(uint16(dev.Port)) |
| 293 | client.SetRetries(dev.Retries) |
| 294 | client.SetTimeout(time.Duration(dev.Timeout) * time.Second) |
| 295 | client.SetMaxOids(dev.MaxOIDs) |
| 296 | client.SetMaxRepetitions(uint32(dev.MaxRepetitions)) |
| 297 | |
| 298 | ver := snmputils.ParseSNMPVersion(dev.SNMPVersion) |
| 299 | |
| 300 | switch ver { |
| 301 | case gosnmp.Version1: |
| 302 | client.SetCommunity(dev.Community) |
| 303 | client.SetVersion(gosnmp.Version1) |
| 304 | case gosnmp.Version2c: |
| 305 | client.SetCommunity(dev.Community) |
| 306 | client.SetVersion(gosnmp.Version2c) |
| 307 | case gosnmp.Version3: |
| 308 | if dev.V3User == "" { |
| 309 | return nil, fmt.Errorf("username is required for SNMPv3") |
| 310 | } |
| 311 | client.SetVersion(gosnmp.Version3) |
| 312 | client.SetSecurityModel(gosnmp.UserSecurityModel) |
| 313 | client.SetMsgFlags(snmputils.ParseSNMPv3SecurityLevel(dev.V3SecurityLevel)) |
| 314 | client.SetSecurityParameters(&gosnmp.UsmSecurityParameters{ |
| 315 | UserName: dev.V3User, |
| 316 | AuthenticationProtocol: snmputils.ParseSNMPv3AuthProtocol(dev.V3AuthProto), |
| 317 | AuthenticationPassphrase: dev.V3AuthKey, |
| 318 | PrivacyProtocol: snmputils.ParseSNMPv3PrivProtocol(dev.V3PrivProto), |
| 319 | PrivacyPassphrase: dev.V3PrivKey, |
| 320 | }) |
| 321 | client.SetContextName(dev.V3ContextName) |
| 322 | default: |
| 323 | return nil, fmt.Errorf("invalid SNMP version: %s", dev.SNMPVersion) |
| 324 | } |
| 325 | |
| 326 | return client, nil |
| 327 | } |
| 328 | |
| 329 | func topologyMethods() []funcapi.MethodConfig { |
| 330 | return []funcapi.MethodConfig{ |
| 331 | topologyMethodConfig(), |
| 332 | } |
| 333 | } |
| 334 | |
| 335 | func topologyFunctionHandler(job collectorapi.RuntimeJob) funcapi.MethodHandler { |
| 336 | return &funcTopology{} |
| 337 | } |