Skip to content

Commit aa7c9a1

Browse files
i3149Copilot
i3149
and
Copilot
authored
Faster api update (#907)
* first pass on a way to use new api to reduce loading time for device apis * Now loading some interfaces too * cleaning up * cleaning up code a bit * Apply suggestions from code review Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * fixing logic on interface lookup --------- Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
1 parent d6ed56b commit aa7c9a1

3 files changed

Lines changed: 760 additions & 87 deletions

File tree

‎pkg/api/api.go‎

Lines changed: 150 additions & 84 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,9 @@ import (
1717
"sync"
1818
"time"
1919

20+
devicepb "github.com/kentik/api-schema-public/gen/go/kentik/device/v202504beta2"
2021
tagging "github.com/kentik/api-schema-public/gen/go/kentik/enrichments/enumerations/v202601alpha1"
22+
interfacepb "github.com/kentik/api-schema-public/gen/go/kentik/interface/v202108alpha1"
2123
synthetics "github.com/kentik/api-schema-public/gen/go/kentik/synthetics/v202309"
2224
"github.com/kentik/ktranslate"
2325
"github.com/kentik/ktranslate/pkg/eggs/logger"
@@ -32,12 +34,14 @@ import (
3234
)
3335

3436
const (
35-
CACHE_TIME_DEVICE = 1 * time.Hour
36-
API_TIMEOUT = 60 * time.Second
37-
HTTP_USER_AGENT = "User-Agent"
38-
API_EMAIL_HEADER = "X-CH-Auth-Email"
39-
API_PASSWORD_HEADER = "X-CH-Auth-API-Token"
40-
MIN_TIME_BETWEEN_SYNTH_CHECKS = 60 * time.Second
37+
CACHE_TIME_DEVICE = 1 * time.Hour
38+
API_TIMEOUT = 60 * time.Second
39+
HTTP_USER_AGENT = "User-Agent"
40+
API_EMAIL_HEADER = "X-CH-Auth-Email"
41+
API_PASSWORD_HEADER = "X-CH-Auth-API-Token"
42+
MIN_TIME_BETWEEN_SYNTH_CHECKS = 60 * time.Second
43+
KT_API_LOAD_INTERFACES = "KT_API_LOAD_INTERFACES"
44+
KT_INTERFACE_LOOKUP_TEXT_FILTER = "KT_INTERFACE_LOOKUP_TEXT_FILTER"
4145
)
4246

4347
var (
@@ -59,10 +63,13 @@ type KentikApi struct {
5963
setTime time.Time
6064
apiTimeout time.Duration
6165
synClient synthetics.SyntheticsAdminServiceClient
66+
deviceClient devicepb.DeviceServiceClient
67+
interfaceClient interfacepb.InterfaceServiceClient
6268
mux sync.RWMutex
6369
lastSynth time.Time
6470
config *ktranslate.Config
6571
tagLookupClient tagging.EnumerationsAdminServiceClient
72+
loadInterfaces bool
6673
}
6774

6875
func NewKentikApi(ctx context.Context, log logger.ContextL, cfg *ktranslate.Config) (*KentikApi, error) {
@@ -85,11 +92,12 @@ func NewKentikApi(ctx context.Context, log logger.ContextL, cfg *ktranslate.Conf
8592
client := &http.Client{Transport: tr, Timeout: apiTimeout}
8693

8794
kapi := &KentikApi{
88-
ContextL: log,
89-
tr: tr,
90-
client: client,
91-
apiTimeout: apiTimeout,
92-
config: cfg,
95+
ContextL: log,
96+
tr: tr,
97+
client: client,
98+
apiTimeout: apiTimeout,
99+
config: cfg,
100+
loadInterfaces: kt.LookupEnvBool(KT_API_LOAD_INTERFACES, false),
93101
}
94102

95103
// Now, check to see if synthetics API works.
@@ -98,16 +106,13 @@ func NewKentikApi(ctx context.Context, log logger.ContextL, cfg *ktranslate.Conf
98106
return nil, err
99107
}
100108

101-
// check api works and also pre-populate cache.
102-
err = kapi.getDevices(ctx)
103-
104109
// Drop cache every hour to keep up to date
105110
go kapi.manageCache(ctx)
106111

107112
return kapi, err
108113
}
109114

110-
func (api *KentikApi) getDeviceInfo(ctx context.Context, apiUrl string, apiEmail string, apiToken string) ([]byte, error) {
115+
func (api *KentikApi) callRestAPI(ctx context.Context, apiUrl string, apiEmail string, apiToken string) ([]byte, error) {
111116
if apiEmail == "" {
112117
if v := api.config.API.DeviceFile; v != "" {
113118
api.Infof("Reading devices from local file: %s", v)
@@ -176,7 +181,7 @@ func (api *KentikApi) UpdateTests(ctx context.Context) {
176181
}
177182

178183
go func() {
179-
err := api.getSynthInfo(ctx)
184+
err := api.getSynthAndDeviceInfo(ctx)
180185
if err != nil {
181186
api.Errorf("Cannot get API synth on demand: %v", err)
182187
}
@@ -238,66 +243,6 @@ func (api *KentikApi) GetDevice(cid kt.Cid, did kt.DeviceID) *kt.Device {
238243
return nil
239244
}
240245

241-
func (api *KentikApi) getInterfaces(ctx context.Context, did kt.DeviceID, info ktranslate.KentikCred) ([]kt.Interface, error) {
242-
res, err := api.getDeviceInfo(ctx, fmt.Sprintf(api.config.APIBaseURL+"/api/internal/device/%d/interfaces", did), info.APIEmail, info.APIToken)
243-
if err != nil {
244-
return nil, err
245-
}
246-
interfaces := []kt.Interface{}
247-
err = json.Unmarshal(res, &interfaces)
248-
return interfaces, err
249-
}
250-
251-
func (api *KentikApi) getDevices(ctx context.Context) error {
252-
stime := time.Now()
253-
resDev := map[kt.Cid]kt.Devices{}
254-
num := 0
255-
for _, info := range api.config.KentikCreds {
256-
res, err := api.getDeviceInfo(ctx, api.config.APIBaseURL+"/api/internal/devices", info.APIEmail, info.APIToken)
257-
if err != nil {
258-
return err
259-
}
260-
var devices kt.DeviceList
261-
err = json.Unmarshal(res, &devices)
262-
if err != nil {
263-
return err
264-
}
265-
266-
for _, device := range devices.Devices {
267-
myd := device
268-
if _, ok := resDev[device.CompanyID]; !ok {
269-
resDev[device.CompanyID] = map[kt.DeviceID]*kt.Device{}
270-
}
271-
myd.Interfaces = map[kt.IfaceID]kt.Interface{}
272-
interfaces, err := api.getInterfaces(ctx, device.ID, info)
273-
if err != nil {
274-
api.Errorf("Cannot get interfaces for %v: %v", device.Name, err)
275-
} else {
276-
for _, intf := range interfaces {
277-
intfl := intf // Should this be a pointer?
278-
myd.Interfaces[intf.SnmpID] = intfl
279-
}
280-
}
281-
resDev[device.CompanyID][device.ID] = &myd
282-
num++
283-
}
284-
285-
api.Infof("Loaded %d Kentik devices via API for %s", len(devices.Devices), info.APIEmail)
286-
}
287-
288-
api.setTime = time.Now()
289-
api.Infof("Loaded %d Kentik devices via API in %v", num, api.setTime.Sub(stime))
290-
api.devices = resDev
291-
292-
// Now pull in site info for these devices.
293-
err := api.getSites(ctx)
294-
if err != nil {
295-
return err
296-
}
297-
298-
return nil
299-
}
300-
301246
func (api *KentikApi) getSites(ctx context.Context) error {
302247
stime := time.Now()
303248
num := 0
@@ -307,7 +252,7 @@ func (api *KentikApi) getSites(ctx context.Context) error {
307252

308253
for _, info := range api.config.KentikCreds {
309254
for _, version := range versions {
310-
res, err := api.getDeviceInfo(ctx, fmt.Sprintf("%s/site/%s/sites", api.config.GRPCBaseURL, version), info.APIEmail, info.APIToken)
255+
res, err := api.callRestAPI(ctx, fmt.Sprintf("%s/site/%s/sites", api.config.GRPCBaseURL, version), info.APIEmail, info.APIToken)
311256
if err != nil {
312257
api.Warnf("Skipping site version %s", version)
313258
continue
@@ -383,11 +328,7 @@ func (api *KentikApi) manageCache(ctx context.Context) {
383328
for {
384329
select {
385330
case _ = <-checkTicker.C:
386-
err := api.getDevices(ctx)
387-
if err != nil {
388-
api.Errorf("Cannot get API devices: %v", err)
389-
}
390-
err = api.getSynthInfo(ctx)
331+
err := api.getSynthAndDeviceInfo(ctx)
391332
if err != nil {
392333
api.Errorf("Cannot get API synth: %v", err)
393334
}
@@ -436,10 +377,12 @@ func (api *KentikApi) connectSynthAndLookup(ctxIn context.Context) error {
436377
}
437378

438379
api.Infof("Connecting to API server at %s", address)
380+
maxMsgSize := 50 * 1024 * 1024 // 50 MB
439381
opts := []grpc.DialOption{
440382
grpc.WithUserAgent(userAgent()),
441383
grpc.WithKeepaliveParams(api.getClientKeepAlive()),
442384
grpc.WithTransportCredentials(creds),
385+
grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMsgSize)),
443386
}
444387

445388
conn, err := grpc.NewClient(address, opts...)
@@ -455,10 +398,20 @@ func (api *KentikApi) connectSynthAndLookup(ctxIn context.Context) error {
455398
api.Infof("Connected to Tag Enum Lookup API server at %s", address)
456399
api.tagLookupClient = clientLookup
457400

458-
return api.getSynthInfo(ctx)
401+
deviceLookup := devicepb.NewDeviceServiceClient(conn)
402+
api.Infof("Connected to Device Lookup API server at %s", address)
403+
api.deviceClient = deviceLookup
404+
405+
if api.loadInterfaces {
406+
interfaceLookup := interfacepb.NewInterfaceServiceClient(conn)
407+
api.Infof("Connected to Interface Lookup API server at %s", address)
408+
api.interfaceClient = interfaceLookup
409+
}
410+
411+
return api.getSynthAndDeviceInfo(ctx)
459412
}
460413

461-
func (api *KentikApi) getSynthInfo(ctx context.Context) error {
414+
func (api *KentikApi) getSynthAndDeviceInfo(ctx context.Context) error {
462415
api.mux.Lock() // Guard this function totally.
463416
defer api.mux.Unlock()
464417
api.lastSynth = time.Now()
@@ -512,6 +465,119 @@ func (api *KentikApi) getSynthInfo(ctx context.Context) error {
512465
api.synTests = synTests
513466
api.Infof("Loaded %d Kentik Tests and %d Agents Total via API", len(api.synTests), len(api.synAgents))
514467

468+
return api.getDeviceInfoNew(ctx)
469+
}
470+
471+
func (api *KentikApi) getDeviceInfoNew(ctx context.Context) error {
472+
473+
stime := time.Now()
474+
resDev := map[kt.Cid]kt.Devices{}
475+
num := 0
476+
deviceIds := []string{}
477+
478+
for _, info := range api.config.KentikCreds {
479+
md := metadata.New(map[string]string{
480+
"X-CH-Auth-Email": info.APIEmail,
481+
"X-CH-Auth-API-Token": info.APIToken,
482+
})
483+
ctxo := metadata.NewOutgoingContext(ctx, md)
484+
485+
lt := &devicepb.ListDevicesRequest{
486+
Query: &devicepb.DeviceQuery{NoCustomColumns: false},
487+
}
488+
r, err := api.deviceClient.ListDevices(ctxo, lt)
489+
if err != nil {
490+
if status.Code(err) == codes.Unimplemented {
491+
api.Warnf("Device ListDevices endpoint not implemented (deprecated API); skipping.")
492+
return nil
493+
}
494+
return err
495+
}
496+
497+
for _, device := range r.GetDevices() {
498+
dev, err := kt.MapDeviceDetailedToDevice(device)
499+
if err != nil {
500+
continue
501+
}
502+
if _, ok := resDev[dev.CompanyID]; !ok {
503+
resDev[dev.CompanyID] = map[kt.DeviceID]*kt.Device{}
504+
}
505+
deviceIds = append(deviceIds, device.Id)
506+
resDev[dev.CompanyID][dev.ID] = dev
507+
num++
508+
}
509+
510+
api.Infof("Loaded %d Kentik Devices via GRPC for %s", len(r.GetDevices()), info.APIEmail)
511+
}
512+
513+
api.devices = resDev
514+
515+
// Now pull in site info for these devices.
516+
err := api.getSites(ctx)
517+
if err != nil {
518+
return err
519+
}
520+
521+
// If we are loading interfaces, pull in interface data now also.
522+
if api.loadInterfaces {
523+
const batchSize = 100
524+
for i := 0; i < len(deviceIds); i += batchSize {
525+
end := min(i+batchSize, len(deviceIds))
526+
err := api.getInterfaces(ctx, deviceIds[i:end])
527+
if err != nil {
528+
return err
529+
}
530+
}
531+
}
532+
533+
api.setTime = time.Now()
534+
api.Infof("Loaded %d Kentik Devices total via GRPC in %v", num, api.setTime.Sub(stime))
535+
return nil
536+
}
537+
538+
func (api *KentikApi) getInterfaces(ctx context.Context, deviceIds []string) error {
539+
for _, info := range api.config.KentikCreds {
540+
api.Infof("Loading %d device interfaces for %s", len(deviceIds), info.APIEmail)
541+
md := metadata.New(map[string]string{
542+
"X-CH-Auth-Email": info.APIEmail,
543+
"X-CH-Auth-API-Token": info.APIToken,
544+
})
545+
ctxo := metadata.NewOutgoingContext(ctx, md)
546+
547+
lt := &interfacepb.ListInterfaceRequest{
548+
Filters: &interfacepb.InterfaceFilter{DeviceIds: deviceIds, Text: kt.LookupEnvString(KT_INTERFACE_LOOKUP_TEXT_FILTER, "")},
549+
}
550+
r, err := api.interfaceClient.ListInterface(ctxo, lt)
551+
if err != nil {
552+
if status.Code(err) == codes.Unimplemented {
553+
api.Warnf("Interface ListInterface endpoint not implemented (deprecated API); skipping.")
554+
return nil
555+
}
556+
return err
557+
}
558+
559+
deviceIndex := make(map[kt.DeviceID]*kt.Device)
560+
for _, clist := range api.devices {
561+
for did, d := range clist {
562+
deviceIndex[did] = d
563+
}
564+
}
565+
566+
found := 0
567+
for _, intf := range r.GetInterfaces() {
568+
deviceID, err := strconv.ParseInt(intf.GetDeviceId(), 10, 64)
569+
if err != nil {
570+
continue
571+
}
572+
if dd, ok := deviceIndex[kt.DeviceID(deviceID)]; ok {
573+
dd.AddInterface(intf)
574+
found++
575+
}
576+
}
577+
578+
api.Infof("Loaded %d Kentik Interfaces via GRPC for %s", found, info.APIEmail)
579+
}
580+
515581
return nil
516582
}
517583

0 commit comments

Comments
 (0)