| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package snmpsd |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "fmt" |
| 8 | "log/slog" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/gohugoio/hashstructure" |
| 12 | "github.com/gosnmp/gosnmp" |
| 13 | "github.com/netdata/netdata/go/plugins/plugin/framework/filepersister" |
| 14 | "github.com/sourcegraph/conc/pool" |
| 15 | |
| 16 | "github.com/netdata/netdata/go/plugins/logger" |
| 17 | "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd/model" |
| 18 | "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/iprange" |
| 19 | "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/snmputils" |
| 20 | ) |
| 21 | |
| 22 | const ( |
| 23 | defaultRescanInterval = time.Minute * 30 |
| 24 | defaultTimeout = time.Second * 1 |
| 25 | defaultParallelScansPerNetwork = 32 |
| 26 | defaultDeviceCacheTTL = time.Hour * 12 |
| 27 | ) |
| 28 | |
| 29 | func NewDiscoverer(cfg Config) (*Discoverer, error) { |
| 30 | subnets, err := cfg.validateAndParse() |
| 31 | if err != nil { |
| 32 | return nil, err |
| 33 | } |
| 34 | |
| 35 | cfgHash, _ := hashstructure.Hash(cfg, nil) |
| 36 | |
| 37 | d := &Discoverer{ |
| 38 | Logger: logger.New().With( |
| 39 | slog.String("component", "service discovery"), |
| 40 | slog.String("discoverer", "snmp"), |
| 41 | ), |
| 42 | cfgSource: cfg.Source, |
| 43 | started: make(chan struct{}), |
| 44 | cfgHash: cfgHash, |
| 45 | subnets: subnets, |
| 46 | newSnmpClient: func() (gosnmp.Handler, func()) { |
| 47 | return gosnmp.NewHandler(), func() {} |
| 48 | }, |
| 49 | |
| 50 | rescanInterval: defaultRescanInterval, |
| 51 | timeout: defaultTimeout, |
| 52 | parallelScansPerNetwork: defaultParallelScansPerNetwork, |
| 53 | deviceCacheTTL: defaultDeviceCacheTTL, |
| 54 | |
| 55 | firstDiscovery: true, |
| 56 | status: newDiscoveryStatus(), |
| 57 | } |
| 58 | |
| 59 | if cfg.RescanInterval.Duration() > 0 { |
| 60 | d.rescanInterval = cfg.RescanInterval.Duration() |
| 61 | } else if cfg.RescanInterval.Duration() < 0 { |
| 62 | d.rescanInterval = 0 // negative means disable rescanning |
| 63 | } |
| 64 | if cfg.Timeout.Duration() > 0 { |
| 65 | d.timeout = cfg.Timeout.Duration() |
| 66 | } |
| 67 | if cfg.ParallelScansPerNetwork > 0 { |
| 68 | d.parallelScansPerNetwork = cfg.ParallelScansPerNetwork |
| 69 | } |
| 70 | if cfg.DeviceCacheTTL.Duration() > 0 { |
| 71 | d.deviceCacheTTL = cfg.DeviceCacheTTL.Duration() |
| 72 | } else if cfg.DeviceCacheTTL.Duration() < 0 { |
| 73 | d.deviceCacheTTL = 0 // negative means cache never expires |
| 74 | } |
| 75 | |
| 76 | return d, nil |
| 77 | } |
| 78 | |
| 79 | type ( |
| 80 | Discoverer struct { |
| 81 | *logger.Logger |
| 82 | model.Base |
| 83 | |
| 84 | cfgSource string // pipeline configuration source |
| 85 | started chan struct{} |
| 86 | cfgHash uint64 |
| 87 | |
| 88 | subnets []subnet |
| 89 | |
| 90 | newSnmpClient func() (gosnmp.Handler, func()) |
| 91 | |
| 92 | parallelScansPerNetwork int |
| 93 | rescanInterval time.Duration |
| 94 | timeout time.Duration |
| 95 | deviceCacheTTL time.Duration |
| 96 | |
| 97 | firstDiscovery bool |
| 98 | status *discoveryStatus |
| 99 | } |
| 100 | subnet struct { |
| 101 | str string |
| 102 | ips iprange.Range |
| 103 | credential CredentialConfig |
| 104 | } |
| 105 | ) |
| 106 | |
| 107 | func (d *Discoverer) String() string { |
| 108 | return "sd:snmp" |
| 109 | } |
| 110 | |
| 111 | func (d *Discoverer) Discover(ctx context.Context, in chan<- []model.TargetGroup) { |
| 112 | d.Info("instance is started") |
| 113 | defer func() { d.Info("instance is stopped") }() |
| 114 | |
| 115 | close(d.started) |
| 116 | |
| 117 | d.loadFileStatus() |
| 118 | |
| 119 | d.discoverNetworks(ctx, in) |
| 120 | |
| 121 | if d.rescanInterval <= 0 { |
| 122 | filepersister.Save(statusFileName(), d.status) |
| 123 | return |
| 124 | } |
| 125 | |
| 126 | tk := time.NewTicker(d.rescanInterval) |
| 127 | defer tk.Stop() |
| 128 | |
| 129 | for { |
| 130 | select { |
| 131 | case <-ctx.Done(): |
| 132 | return |
| 133 | case <-tk.C: |
| 134 | d.discoverNetworks(ctx, in) |
| 135 | } |
| 136 | } |
| 137 | } |
| 138 | |
| 139 | func (d *Discoverer) discoverNetworks(ctx context.Context, in chan<- []model.TargetGroup) { |
| 140 | now := time.Now() |
| 141 | |
| 142 | doProbing := !d.firstDiscovery || |
| 143 | d.status.ConfigHash != d.cfgHash || |
| 144 | now.After(d.status.LastDiscoveryTime.Add(d.rescanInterval)) |
| 145 | |
| 146 | defer func() { |
| 147 | if isDone(ctx) { |
| 148 | return |
| 149 | } |
| 150 | d.firstDiscovery = false |
| 151 | |
| 152 | if doProbing { |
| 153 | d.status.LastDiscoveryTime = now |
| 154 | } |
| 155 | |
| 156 | if d.status.updated.Swap(false) || d.status.ConfigHash != d.cfgHash { |
| 157 | d.status.ConfigHash = d.cfgHash |
| 158 | filepersister.Save(statusFileName(), d.status) |
| 159 | } |
| 160 | }() |
| 161 | |
| 162 | d.Infof("discovery mode: %s", map[bool]string{true: "active probing", false: "using cache"}[doProbing]) |
| 163 | |
| 164 | p := pool.New() |
| 165 | for _, sub := range d.subnets { |
| 166 | p.Go(func() { d.discoverNetwork(ctx, in, sub, doProbing) }) |
| 167 | } |
| 168 | p.Wait() |
| 169 | } |
| 170 | |
| 171 | func (d *Discoverer) discoverNetwork(ctx context.Context, in chan<- []model.TargetGroup, sub subnet, doProbing bool) { |
| 172 | tgg := newTargetGroup(sub) |
| 173 | if d.cfgSource != "" { |
| 174 | tgg.source += fmt.Sprintf(",%s", d.cfgSource) |
| 175 | } |
| 176 | p := pool.New().WithMaxGoroutines(d.parallelScansPerNetwork) |
| 177 | |
| 178 | client, cleanup := d.newSnmpClient() |
| 179 | defer cleanup() |
| 180 | |
| 181 | client.SetTimeout(d.timeout) |
| 182 | client.SetRetries(0) |
| 183 | setCredential(client, sub.credential) |
| 184 | d.Debugf("SNMP client info for '%s': %s", sub.str, snmputils.SnmpClientConnInfo(client)) |
| 185 | |
| 186 | for ip := range sub.ips.Iterate() { |
| 187 | ipAddr := ip.String() |
| 188 | |
| 189 | if doProbing { |
| 190 | p.Go(func() { d.probeIPAddress(ctx, sub, ipAddr, tgg) }) |
| 191 | } else { |
| 192 | d.useCacheIPAddress(sub, ipAddr, tgg) |
| 193 | } |
| 194 | } |
| 195 | p.Wait() |
| 196 | |
| 197 | send(ctx, in, tgg) |
| 198 | } |
| 199 | |
| 200 | func (d *Discoverer) useCacheIPAddress(sub subnet, ip string, tgg *targetGroup) { |
| 201 | if dev := d.status.get(sub, ip); dev != nil { |
| 202 | tg := newTarget(ip, sub.credential, dev.SysInfo) |
| 203 | tgg.addTarget(tg) |
| 204 | } |
| 205 | } |
| 206 | |
| 207 | func (d *Discoverer) probeIPAddress(ctx context.Context, sub subnet, ip string, tgg *targetGroup) { |
| 208 | if isDone(ctx) { |
| 209 | return |
| 210 | } |
| 211 | |
| 212 | now := time.Now() |
| 213 | |
| 214 | dev := d.status.get(sub, ip) |
| 215 | |
| 216 | // Use the cached device if available and not expired |
| 217 | if dev != nil && (d.deviceCacheTTL == 0 || now.Before(dev.DiscoverTime.Add(d.deviceCacheTTL))) { |
| 218 | if d.firstDiscovery { |
| 219 | if d.deviceCacheTTL == 0 { |
| 220 | d.Infof("device '%s': found in cache (sysName: '%s', network: '%s', cache never expires)", |
| 221 | ip, dev.SysInfo.Name, subKey(sub)) |
| 222 | } else { |
| 223 | untilProbe := dev.DiscoverTime.Add(d.deviceCacheTTL).Sub(now).Round(time.Second) |
| 224 | d.Infof("device '%s': found in cache (sysName: '%s', network: '%s', next probe in %s)", |
| 225 | ip, dev.SysInfo.Name, subKey(sub), untilProbe) |
| 226 | } |
| 227 | } |
| 228 | tg := newTarget(ip, sub.credential, dev.SysInfo) |
| 229 | tgg.addTarget(tg) |
| 230 | return |
| 231 | } |
| 232 | |
| 233 | si, err := d.getSnmpSysInfo(sub, ip) |
| 234 | if err != nil { |
| 235 | if dev == nil { |
| 236 | // First-time discovery failure - log at debug level as this is expected for many IPs |
| 237 | d.Debugf("device '%s': probe failed (network: '%s'): %v", ip, subKey(sub), err) |
| 238 | } else { |
| 239 | // Previously discovered device is now unreachable |
| 240 | d.Warningf("lost connection to previously discovered SNMP device '%s' (sysName: '%s', network: '%s'): %v", |
| 241 | ip, dev.SysInfo.Name, subKey(sub), err) |
| 242 | } |
| 243 | d.status.del(sub, ip) |
| 244 | d.status.updated.Store(dev != nil) |
| 245 | return |
| 246 | } |
| 247 | |
| 248 | d.Infof("device '%s': successfully discovered (sysName: '%s', network: '%s')", ip, si.Name, subKey(sub)) |
| 249 | d.status.put(sub, ip, &discoveredDevice{DiscoverTime: now, SysInfo: *si}) |
| 250 | d.status.updated.Store(true) |
| 251 | tg := newTarget(ip, sub.credential, *si) |
| 252 | tgg.addTarget(tg) |
| 253 | } |
| 254 | |
| 255 | func (d *Discoverer) getSnmpSysInfo(sub subnet, ip string) (*snmputils.SysInfo, error) { |
| 256 | client, cleanup := d.newSnmpClient() |
| 257 | defer cleanup() |
| 258 | |
| 259 | client.SetTarget(ip) |
| 260 | client.SetTimeout(d.timeout) |
| 261 | client.SetRetries(0) |
| 262 | setCredential(client, sub.credential) |
| 263 | |
| 264 | if err := client.Connect(); err != nil { |
| 265 | return nil, fmt.Errorf("failed to connect: %v", err) |
| 266 | } |
| 267 | |
| 268 | defer func() { _ = client.Close() }() |
| 269 | |
| 270 | return snmputils.GetSysInfo(client) |
| 271 | } |
| 272 | |
| 273 | func send(ctx context.Context, in chan<- []model.TargetGroup, tgg model.TargetGroup) { |
| 274 | select { |
| 275 | case <-ctx.Done(): |
| 276 | case in <- []model.TargetGroup{tgg}: |
| 277 | } |
| 278 | } |
| 279 | func isDone(ctx context.Context) bool { |
| 280 | select { |
| 281 | case <-ctx.Done(): |
| 282 | return true |
| 283 | default: |
| 284 | return false |
| 285 | } |
| 286 | } |