Skip to content

Commit 7f41c23

Browse files
i3149i3149
authored
Allowing for lazy loading of interfaces (#914)
* Allowing for lazy loading of interfaces * Better interface loading code which simplifies the lookups * Guards on interface loading
1 parent 5900163 commit 7f41c23

2 files changed

Lines changed: 82 additions & 104 deletions

File tree

‎pkg/api/api.go‎

Lines changed: 50 additions & 80 deletions
Original file line numberDiff line numberDiff line change
@@ -40,10 +40,10 @@ const (
4040
API_EMAIL_HEADER = "X-CH-Auth-Email"
4141
API_PASSWORD_HEADER = "X-CH-Auth-API-Token"
4242
MIN_TIME_BETWEEN_SYNTH_CHECKS = 60 * time.Second
43-
KT_API_LOAD_INTERFACES = "KT_API_LOAD_INTERFACES"
4443
KT_INTERFACE_LOOKUP_TEXT_FILTER = "KT_INTERFACE_LOOKUP_TEXT_FILTER"
4544
KT_LOAD_CUSTOM_COLUMNS = "KT_LOAD_CUSTOM_COLUMNS"
4645
KT_API_LAZY_LOAD_CUSTOMS = "KT_API_LAZY_LOAD_CUSTOMS"
46+
KT_API_LAZY_LOAD_INTERFACES = "KT_API_LAZY_LOAD_INTERFACES"
4747
)
4848

4949
var (
@@ -56,24 +56,25 @@ func init() {
5656

5757
type KentikApi struct {
5858
logger.ContextL
59-
tr *http.Transport
60-
client *http.Client
61-
devices map[kt.Cid]kt.Devices
62-
synAgents map[kt.AgentId]*synthetics.Agent
63-
synAgentsByIP map[string]*synthetics.Agent
64-
synTests map[kt.TestId]*synthetics.Test
65-
setTime time.Time
66-
apiTimeout time.Duration
67-
synClient synthetics.SyntheticsAdminServiceClient
68-
deviceClient devicepb.DeviceServiceClient
69-
interfaceClient interfacepb.InterfaceServiceClient
70-
mux sync.RWMutex
71-
lastSynth time.Time
72-
config *ktranslate.Config
73-
tagLookupClient tagging.EnumerationsAdminServiceClient
74-
loadInterfaces bool
75-
lazyLoadCustoms bool
76-
fullLoadCustoms bool
59+
tr *http.Transport
60+
client *http.Client
61+
devices map[kt.Cid]kt.Devices
62+
synAgents map[kt.AgentId]*synthetics.Agent
63+
synAgentsByIP map[string]*synthetics.Agent
64+
synTests map[kt.TestId]*synthetics.Test
65+
setTime time.Time
66+
apiTimeout time.Duration
67+
synClient synthetics.SyntheticsAdminServiceClient
68+
deviceClient devicepb.DeviceServiceClient
69+
interfaceClient interfacepb.InterfaceServiceClient
70+
mux sync.RWMutex
71+
lastSynth time.Time
72+
config *ktranslate.Config
73+
tagLookupClient tagging.EnumerationsAdminServiceClient
74+
lazyLoadCustoms bool
75+
fullLoadCustoms bool
76+
lazyLoadInts bool
77+
interfaceFilterText string
7778
}
7879

7980
func NewKentikApi(ctx context.Context, log logger.ContextL, cfg *ktranslate.Config) (*KentikApi, error) {
@@ -95,17 +96,19 @@ func NewKentikApi(ctx context.Context, log logger.ContextL, cfg *ktranslate.Conf
9596
client := &http.Client{Transport: tr, Timeout: apiTimeout}
9697

9798
kapi := &KentikApi{
98-
ContextL: log,
99-
tr: tr,
100-
client: client,
101-
apiTimeout: apiTimeout,
102-
config: cfg,
103-
loadInterfaces: kt.LookupEnvBool(KT_API_LOAD_INTERFACES, false),
104-
lazyLoadCustoms: kt.LookupEnvBool(KT_API_LAZY_LOAD_CUSTOMS, false),
105-
fullLoadCustoms: kt.LookupEnvBool(KT_LOAD_CUSTOM_COLUMNS, false),
99+
ContextL: log,
100+
tr: tr,
101+
client: client,
102+
apiTimeout: apiTimeout,
103+
config: cfg,
104+
lazyLoadCustoms: kt.LookupEnvBool(KT_API_LAZY_LOAD_CUSTOMS, false),
105+
fullLoadCustoms: kt.LookupEnvBool(KT_LOAD_CUSTOM_COLUMNS, false),
106+
lazyLoadInts: kt.LookupEnvBool(KT_API_LAZY_LOAD_INTERFACES, false),
107+
interfaceFilterText: kt.LookupEnvString(KT_INTERFACE_LOOKUP_TEXT_FILTER, ""),
106108
}
107109

108-
log.Infof("Setting API timeout to %v, loadInterfaces=%v, lazyLoadCustoms=%v, fullyLoadCustoms=%v", apiTimeout, kapi.loadInterfaces, kapi.lazyLoadCustoms, kapi.fullLoadCustoms)
110+
log.Infof("Setting API timeout to %v, lazyLoadInterfaces=%v, lazyLoadCustoms=%v, fullyLoadCustoms=%v",
111+
apiTimeout, kapi.lazyLoadInts, kapi.lazyLoadCustoms, kapi.fullLoadCustoms)
109112

110113
// Now, check to see if synthetics API works.
111114
err := kapi.connectSynthAndLookup(ctx)
@@ -261,6 +264,12 @@ func (api *KentikApi) GetDevice(ctx context.Context, cid kt.Cid, did kt.DeviceID
261264
if err != nil {
262265
api.Warnf("Cannot load customs for %s %s, %v", dev.IDStr, info.APIEmail, err)
263266
} else {
267+
if api.lazyLoadInts {
268+
err := api.getInterfaces(ctxo, dev.IDStr, dev)
269+
if err != nil {
270+
api.Warnf("Cannot load interfaces for %s %s, %v", dev.IDStr, info.APIEmail, err)
271+
}
272+
}
264273
return
265274
}
266275
}
@@ -430,7 +439,7 @@ func (api *KentikApi) connectSynthAndLookup(ctxIn context.Context) error {
430439
api.Infof("Connected to Device Lookup API server at %s", address)
431440
api.deviceClient = deviceLookup
432441

433-
if api.loadInterfaces {
442+
if api.lazyLoadInts {
434443
interfaceLookup := interfacepb.NewInterfaceServiceClient(conn)
435444
api.Infof("Connected to Interface Lookup API server at %s", address)
436445
api.interfaceClient = interfaceLookup
@@ -568,66 +577,27 @@ func (api *KentikApi) getDeviceInfoNew(ctx context.Context) error {
568577
return err
569578
}
570579

571-
// If we are loading interfaces, pull in interface data now also.
572-
if api.loadInterfaces {
573-
const batchSize = 100
574-
for i := 0; i < len(deviceIds); i += batchSize {
575-
end := min(i+batchSize, len(deviceIds))
576-
err := api.getInterfaces(ctx, deviceIds[i:end])
577-
if err != nil {
578-
return err
579-
}
580-
}
581-
}
582-
583580
api.setTime = time.Now()
584581
api.Infof("Loaded %d Kentik Devices total via GRPC in %v", num, api.setTime.Sub(stime))
585582
return nil
586583
}
587584

588-
func (api *KentikApi) getInterfaces(ctx context.Context, deviceIds []string) error {
589-
for _, info := range api.config.KentikCreds {
590-
api.Infof("Loading %d device interfaces for %s", len(deviceIds), info.APIEmail)
591-
md := metadata.New(map[string]string{
592-
"X-CH-Auth-Email": info.APIEmail,
593-
"X-CH-Auth-API-Token": info.APIToken,
594-
})
595-
ctxo := metadata.NewOutgoingContext(ctx, md)
596-
597-
lt := &interfacepb.ListInterfaceRequest{
598-
Filters: &interfacepb.InterfaceFilter{DeviceIds: deviceIds, Text: kt.LookupEnvString(KT_INTERFACE_LOOKUP_TEXT_FILTER, "")},
599-
}
600-
r, err := api.interfaceClient.ListInterface(ctxo, lt)
601-
if err != nil {
602-
if status.Code(err) == codes.Unimplemented {
603-
api.Warnf("Interface ListInterface endpoint not implemented (deprecated API); skipping.")
604-
return nil
605-
}
606-
return err
607-
}
608-
609-
deviceIndex := make(map[kt.DeviceID]*kt.Device)
610-
for _, clist := range api.devices {
611-
for did, d := range clist {
612-
deviceIndex[did] = d
613-
}
614-
}
585+
func (api *KentikApi) getInterfaces(ctx context.Context, deviceId string, dev *kt.Device) error {
615586

616-
found := 0
617-
for _, intf := range r.GetInterfaces() {
618-
deviceID, err := strconv.ParseInt(intf.GetDeviceId(), 10, 64)
619-
if err != nil {
620-
continue
621-
}
622-
if dd, ok := deviceIndex[kt.DeviceID(deviceID)]; ok {
623-
dd.AddInterface(intf)
624-
found++
625-
}
587+
api.Debugf("Loading interfaces for %s", deviceId)
588+
lt := &interfacepb.ListInterfaceRequest{
589+
Filters: &interfacepb.InterfaceFilter{DeviceIds: []string{deviceId}, Text: api.interfaceFilterText},
590+
}
591+
r, err := api.interfaceClient.ListInterface(ctx, lt)
592+
if err != nil {
593+
if status.Code(err) == codes.Unimplemented {
594+
api.Warnf("Interface ListInterface endpoint not implemented (deprecated API); skipping.")
595+
return nil
626596
}
627-
628-
api.Infof("Loaded %d Kentik Interfaces via GRPC for %s", found, info.APIEmail)
597+
return err
629598
}
630599

600+
dev.AddInterfaces(r.GetInterfaces())
631601
return nil
632602
}
633603

‎pkg/kt/device_types.go‎

Lines changed: 32 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -184,35 +184,43 @@ func (d *Device) LoadedCustoms() bool {
184184
return lc
185185
}
186186

187-
func (d *Device) AddInterface(p *interfacepb.Interface) {
188-
if p == nil {
189-
return
190-
}
187+
func (d *Device) AddInterfaces(ints []*interfacepb.Interface) {
191188

192-
snmpID, err := strconv.ParseInt(p.GetSnmpId(), 10, 64)
193-
if err != nil {
194-
snmpID = 0
195-
}
189+
ifaces := make([]Interface, len(ints))
190+
ifacemap := map[IfaceID]Interface{}
191+
for i, p := range ints {
192+
if p == nil {
193+
continue
194+
}
196195

197-
devID, err := strconv.ParseInt(p.GetDeviceId(), 10, 64)
198-
if err != nil {
199-
devID = 0
200-
}
196+
snmpID, err := strconv.ParseInt(p.GetSnmpId(), 10, 64)
197+
if err != nil {
198+
snmpID = 0
199+
}
200+
201+
devID, err := strconv.ParseInt(p.GetDeviceId(), 10, 64)
202+
if err != nil {
203+
devID = 0
204+
}
205+
206+
iface := Interface{
207+
DeviceID: DeviceID(devID),
208+
Description: p.GetInterfaceDescription(),
209+
NetworkBoundary: ic.NameFromNBInt(int(p.GetNetworkBoundary())),
210+
ConnectivityType: ic.NameFromCTInt(int(p.GetConnectivityType())),
211+
Provider: p.GetProvider(),
212+
SnmpID: IfaceID(snmpID),
213+
Alias: p.GetSnmpAlias(),
214+
SnmpSpeedMbps: int64(p.GetSnmpSpeed()),
215+
Address: p.GetInterfaceIp(),
216+
}
201217

202-
iface := Interface{
203-
DeviceID: DeviceID(devID),
204-
Description: p.GetInterfaceDescription(),
205-
NetworkBoundary: ic.NameFromNBInt(int(p.GetNetworkBoundary())),
206-
ConnectivityType: ic.NameFromCTInt(int(p.GetConnectivityType())),
207-
Provider: p.GetProvider(),
208-
SnmpID: IfaceID(snmpID),
209-
Alias: p.GetSnmpAlias(),
210-
SnmpSpeedMbps: int64(p.GetSnmpSpeed()),
211-
Address: p.GetInterfaceIp(),
218+
ifaces[i] = iface
219+
ifacemap[IfaceID(snmpID)] = iface
212220
}
213221

214-
d.AllInterfaces = append(d.AllInterfaces, iface)
215-
d.Interfaces[IfaceID(snmpID)] = iface
222+
d.AllInterfaces = ifaces
223+
d.Interfaces = ifacemap
216224
}
217225

218226
func MapDeviceDetailedToDevice(dd *devicepb.DeviceDetailed) (*Device, error) {

0 commit comments

Comments
 (0)