989241608

Committed 01 Sep 2023 02:07PM UTC coverage: 91.613% (+0.6%) from 91.042%

Build # 989241608

Build Type

Pull #409

gitlab-ci

Committed by

tranchitella

Commit Message

feat: update device inventory only when changed but at least daily

Changelog: Title
Ticket: MEN-6643

Signed-off-by: Fabio Tranchitella <fabio.tranchitella@northern.tech>

Pull Request Pull Request #409: feat: update device inventory only when changed but at least daily

Run Details

56 of 58 new or added lines in 2 files covered. (96.55%)

1 existing line in 1 file now uncovered.

3135 of 3422 relevant lines covered (91.61%)

140.58 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.69

/inv/inventory.go

// Copyright 2023 Northern.tech AS
//
//    Licensed under the Apache License, Version 2.0 (the "License");
//    you may not use this file except in compliance with the License.
//    You may obtain a copy of the License at
//
//        http://www.apache.org/licenses/LICENSE-2.0
//
//    Unless required by applicable law or agreed to in writing, software
//    distributed under the License is distributed on an "AS IS" BASIS,
//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//    See the License for the specific language governing permissions and
//    limitations under the License.

package inv

import (
        "context"
        "reflect"
        "time"

        "github.com/pkg/errors"

        "github.com/mendersoftware/go-lib-micro/log"

        "github.com/mendersoftware/inventory/client/devicemonitor"
        "github.com/mendersoftware/inventory/client/workflows"
        "github.com/mendersoftware/inventory/model"
        "github.com/mendersoftware/inventory/store"
        "github.com/mendersoftware/inventory/store/mongo"
        "github.com/mendersoftware/inventory/utils"
)

const reindexBatchSize = 100

var (
        ErrETagDoesntMatch   = errors.New("ETag does not match")
        ErrTooManyAttributes = errors.New("the number of attributes in the scope is above the limit")
)

// this inventory service interface
//
//go:generate ../utils/mockgen.sh
type InventoryApp interface {
        WithReporting(c workflows.Client) InventoryApp
        HealthCheck(ctx context.Context) error
        ListDevices(ctx context.Context, q store.ListQuery) ([]model.Device, int, error)
        GetDevice(ctx context.Context, id model.DeviceID) (*model.Device, error)
        AddDevice(ctx context.Context, d *model.Device) error
        UpsertAttributes(ctx context.Context, id model.DeviceID, attrs model.DeviceAttributes) error
        UpsertAttributesWithUpdated(
                ctx context.Context,
                id model.DeviceID,
                attrs model.DeviceAttributes,
                scope string,
                etag string,
        ) error
        UpsertDevicesStatuses(
                ctx context.Context,
                devices []model.DeviceUpdate,
                attrs model.DeviceAttributes,
        ) (*model.UpdateResult, error)
        ReplaceAttributes(
                ctx context.Context,
                id model.DeviceID,
                upsertAttrs model.DeviceAttributes,
                scope string,
                etag string,
        ) error
        GetFiltersAttributes(ctx context.Context) ([]model.FilterAttribute, error)
        DeleteGroup(ctx context.Context, groupName model.GroupName) (*model.UpdateResult, error)
        UnsetDeviceGroup(ctx context.Context, id model.DeviceID, groupName model.GroupName) error
        UnsetDevicesGroup(
                ctx context.Context,
                deviceIDs []model.DeviceID,
                groupName model.GroupName,
        ) (*model.UpdateResult, error)
        UpdateDeviceGroup(ctx context.Context, id model.DeviceID, group model.GroupName) error
        UpdateDevicesGroup(
                ctx context.Context,
                ids []model.DeviceID,
                group model.GroupName,
        ) (*model.UpdateResult, error)
        ListGroups(ctx context.Context, filters []model.FilterPredicate) ([]model.GroupName, error)
        ListDevicesByGroup(
                ctx context.Context,
                group model.GroupName,
                skip int,
                limit int,
        ) ([]model.DeviceID, int, error)
        GetDeviceGroup(ctx context.Context, id model.DeviceID) (model.GroupName, error)
        DeleteDevice(ctx context.Context, id model.DeviceID) error
        DeleteDevices(
                ctx context.Context,
                ids []model.DeviceID,
        ) (*model.UpdateResult, error)
        CreateTenant(ctx context.Context, tenant model.NewTenant) error
        SearchDevices(ctx context.Context, searchParams model.SearchParams) ([]model.Device, int, error)
        CheckAlerts(ctx context.Context, deviceId string) (int, error)
        WithLimits(attributes, tags int) InventoryApp
        WithDevicemonitor(client devicemonitor.Client) InventoryApp
}

type inventory struct {
        db              store.DataStore
        limitAttributes int
        limitTags       int
        dmClient        devicemonitor.Client
        enableReporting bool
        wfClient        workflows.Client
}

func NewInventory(d store.DataStore) InventoryApp {
        return &inventory{db: d}
}

func (i *inventory) WithDevicemonitor(client devicemonitor.Client) InventoryApp {
        i.dmClient = client
        return i
}

func (i *inventory) WithLimits(limitAttributes, limitTags int) InventoryApp {
        i.limitAttributes = limitAttributes
        i.limitTags = limitTags
        return i
}

func (i *inventory) WithReporting(client workflows.Client) InventoryApp {
        i.enableReporting = true
        i.wfClient = client
        return i
}

func (i *inventory) HealthCheck(ctx context.Context) error {
        err := i.db.Ping(ctx)
        if err != nil {
                return errors.Wrap(err, "error reaching MongoDB")
        }

        if i.enableReporting {
                err := i.wfClient.CheckHealth(ctx)
                if err != nil {
                        return errors.Wrap(err, "error reaching workflows")
                }
        }

        return nil
}

func (i *inventory) ListDevices(
        ctx context.Context,
        q store.ListQuery,
) ([]model.Device, int, error) {
        devs, totalCount, err := i.db.GetDevices(ctx, q)

        if err != nil {
                return nil, -1, errors.Wrap(err, "failed to fetch devices")
        }

        return devs, totalCount, nil
}

func (i *inventory) GetDevice(ctx context.Context, id model.DeviceID) (*model.Device, error) {
        dev, err := i.db.GetDevice(ctx, id)
        if err != nil {
                return nil, errors.Wrap(err, "failed to fetch device")
        }
        return dev, nil
}

func (i *inventory) AddDevice(ctx context.Context, dev *model.Device) error {
        if dev == nil {
                return errors.New("no device given")
        }
        dev.Text = utils.GetTextField(dev)
        err := i.db.AddDevice(ctx, dev)
        if err != nil {
                return errors.Wrap(err, "failed to add device")
        }

        i.maybeTriggerReindex(ctx, []model.DeviceID{dev.ID})

        return nil
}

func (i *inventory) DeleteDevices(
        ctx context.Context,
        ids []model.DeviceID,
) (*model.UpdateResult, error) {
        res, err := i.db.DeleteDevices(ctx, ids)
        if err != nil {
                return nil, err
        }

        if i.enableReporting {
                for _, d := range ids {
                        i.triggerReindex(ctx, []model.DeviceID{d})
                }
        }

        return res, err
}

func (i *inventory) DeleteDevice(ctx context.Context, id model.DeviceID) error {
        res, err := i.db.DeleteDevices(ctx, []model.DeviceID{id})
        if err != nil {
                return errors.Wrap(err, "failed to delete device")
        } else if res.DeletedCount < 1 {
                return store.ErrDevNotFound
        }
        i.maybeTriggerReindex(ctx, []model.DeviceID{id})

        return nil
}

func (i *inventory) UpsertAttributes(
        ctx context.Context,
        id model.DeviceID,
        attrs model.DeviceAttributes,
) error {
        res, err := i.db.UpsertDevicesAttributes(
                ctx, []model.DeviceID{id}, attrs,
        )
        if err != nil {
                return errors.Wrap(err, "failed to upsert attributes in db")
        }
        if res != nil && res.MatchedCount > 0 {
                i.reindexTextField(ctx, res.Devices)
                i.maybeTriggerReindex(ctx, []model.DeviceID{id})
        }
        return nil
}

func (i *inventory) checkAttributesLimits(
        ctx context.Context,
        id model.DeviceID,
        attrs model.DeviceAttributes,
        scope string,
) error {
        limit := 0
        switch scope {
        case model.AttrScopeInventory:
                limit = i.limitAttributes
        case model.AttrScopeTags:
                limit = i.limitTags
        }
        if limit == 0 {
                return nil
        }
        device, err := i.db.GetDevice(ctx, id)
        if err != nil && err != store.ErrDevNotFound {
                return errors.Wrap(err, "failed to get the device")
        } else if device == nil {
                return nil
        }
        count := 0
        for _, attr := range device.Attributes {
                if attr.Scope == scope {
                        count += 1
                        if count > limit {
                                break
                        }
                }
        }
        for _, attr := range attrs {
                if count > limit {
                        break
                }
                found := false
                for _, devAttr := range device.Attributes {
                        if attr.Scope == scope && attr.Name == devAttr.Name {
                                found = true
                        }
                }
                if !found {
                        count++
                }
        }
        if count > limit {
                return ErrTooManyAttributes
        }
        return nil
}

const oneDay = 24 * time.Hour

func (i *inventory) needsUpsert(
        device *model.Device,
        upsertAttrs model.DeviceAttributes,
        removeAttrs model.DeviceAttributes,
) bool {
        needsUpsert := true
        if device != nil {
                // we update the inventory attributes at least once every (calendar) day
                if !device.UpdatedTs.IsZero() &&
                        device.UpdatedTs.Truncate(oneDay) != time.Now().Truncate(oneDay) {
                        return true
                }
                needsUpsert = false
                for _, attribute := range upsertAttrs {
                        if attribute.Scope != model.AttrScopeInventory {
                                needsUpsert = true
                                break
                        }
                        attributeChanged := true
                        for _, deviceAttribute := range device.Attributes {
                                if !(attribute.Scope == deviceAttribute.Scope &&
                                        attribute.Name == deviceAttribute.Name) {
                                        continue
                                }
                                attributeChanged = !reflect.DeepEqual(attribute.Value, deviceAttribute.Value)
                                break
                        }
                        if attributeChanged {
                                needsUpsert = true
                                break
                        }
                }
                if !needsUpsert && len(removeAttrs) > 0 {
                OuterLoop:
                        for _, attribute := range removeAttrs {
                                for _, deviceAttribute := range device.Attributes {
                                        if attribute.Scope == deviceAttribute.Scope &&
                                                attribute.Name == deviceAttribute.Name {
                                                needsUpsert = true
                                                break OuterLoop
                                        }
                                }
                        }
                }
        }
        return needsUpsert
}

func (i *inventory) UpsertAttributesWithUpdated(
        ctx context.Context,
        id model.DeviceID,
        attrs model.DeviceAttributes,
        scope string,
        etag string,
) error {
        if err := i.checkAttributesLimits(ctx, id, attrs, scope); err != nil {
                return err
        }

        device, err := i.db.GetDevice(ctx, id)
        if err != nil && err != store.ErrDevNotFound {
                return errors.Wrap(err, "failed to get the device")
        } else if !i.needsUpsert(device, attrs, nil) {
                return nil
        }

        res, err := i.db.UpsertDevicesAttributesWithUpdated(
                ctx, []model.DeviceID{id}, attrs, scope, etag,
        )
        if err != nil {
                return errors.Wrap(err, "failed to upsert attributes in db")
        }
        if scope == model.AttrScopeTags {
                if res != nil && res.MatchedCount == 0 && etag != "" {
                        return ErrETagDoesntMatch
                }
        }

        if res != nil && res.MatchedCount > 0 {
                i.reindexTextField(ctx, res.Devices)
                i.maybeTriggerReindex(ctx, []model.DeviceID{id})
        }
        return nil
}

func getRemoveAttrs(
        device *model.Device,
        scope string,
        upsertAttrs model.DeviceAttributes,
) model.DeviceAttributes {
        removeAttrs := model.DeviceAttributes{}
        if device != nil {
                for _, attr := range device.Attributes {
                        if attr.Scope == scope {
                                update := false
                                for _, upsertAttr := range upsertAttrs {
                                        if upsertAttr.Name == attr.Name {
                                                update = true
                                                break
                                        }
                                }
                                if !update {
                                        removeAttrs = append(removeAttrs, attr)
                                }
                        }
                }
        }
        return removeAttrs
}

func (i *inventory) ReplaceAttributes(
        ctx context.Context,
        id model.DeviceID,
        upsertAttrs model.DeviceAttributes,
        scope string,
        etag string,
) error {
        limit := 0
        switch scope {
        case model.AttrScopeInventory:
                limit = i.limitAttributes
        case model.AttrScopeTags:
                limit = i.limitTags
        }
        if limit > 0 && len(upsertAttrs) > limit {
                return ErrTooManyAttributes
        }

        device, err := i.db.GetDevice(ctx, id)
        if err != nil && err != store.ErrDevNotFound {
                return errors.Wrap(err, "failed to get the device")
        }

        removeAttrs := getRemoveAttrs(device, scope, upsertAttrs)
        if !i.needsUpsert(device, upsertAttrs, removeAttrs) {
                return nil
        }

        res, err := i.db.UpsertRemoveDeviceAttributes(ctx, id, upsertAttrs, removeAttrs, scope, etag)
        if err != nil {
                return errors.Wrap(err, "failed to replace attributes in db")
        }
        if scope == model.AttrScopeTags {
                if res != nil && res.MatchedCount == 0 && etag != "" {
                        return ErrETagDoesntMatch
                }
        }
        if res != nil && res.MatchedCount > 0 {
                i.reindexTextField(ctx, res.Devices)
                i.maybeTriggerReindex(ctx, []model.DeviceID{id})
        }
        return nil
}

func (i *inventory) GetFiltersAttributes(ctx context.Context) ([]model.FilterAttribute, error) {
        attributes, err := i.db.GetFiltersAttributes(ctx)
        if err != nil {
                return nil, errors.Wrap(err, "failed to get filter attributes from the db")
        }
        return attributes, nil
}

func (i *inventory) DeleteGroup(
        ctx context.Context,
        groupName model.GroupName,
) (*model.UpdateResult, error) {
        deviceIDs, err := i.db.DeleteGroup(ctx, groupName)
        if err != nil {
                return nil, errors.Wrap(err, "failed to delete group")
        }

        batchDeviceIDsLength := 0
        batchDeviceIDs := make([]model.DeviceID, reindexBatchSize)

        triggerReindex := func() {
                i.maybeTriggerReindex(ctx, batchDeviceIDs[0:batchDeviceIDsLength])
                batchDeviceIDsLength = 0
        }

        res := &model.UpdateResult{}
        for deviceID := range deviceIDs {
                batchDeviceIDs[batchDeviceIDsLength] = deviceID
                batchDeviceIDsLength++
                if batchDeviceIDsLength == reindexBatchSize {
                        triggerReindex()
                }
                res.MatchedCount += 1
                res.UpdatedCount += 1
        }
        if batchDeviceIDsLength > 0 {
                triggerReindex()
        }

        return res, err
}

func (i *inventory) UpsertDevicesStatuses(
        ctx context.Context,
        devices []model.DeviceUpdate,
        attrs model.DeviceAttributes,
) (*model.UpdateResult, error) {
        res, err := i.db.UpsertDevicesAttributesWithRevision(ctx, devices, attrs)
        if err != nil {
                return nil, err
        }

        if i.enableReporting {
                deviceIDs := make([]model.DeviceID, len(devices))
                for i, d := range devices {
                        deviceIDs[i] = d.Id
                }
                i.triggerReindex(ctx, deviceIDs)
        }

        return res, err
}

func (i *inventory) UnsetDevicesGroup(
        ctx context.Context,
        deviceIDs []model.DeviceID,
        groupName model.GroupName,
) (*model.UpdateResult, error) {
        res, err := i.db.UnsetDevicesGroup(ctx, deviceIDs, groupName)
        if err != nil {
                return nil, err
        }

        if i.enableReporting {
                i.triggerReindex(ctx, deviceIDs)
        }

        return res, nil
}

func (i *inventory) UnsetDeviceGroup(
        ctx context.Context,
        id model.DeviceID,
        group model.GroupName,
) error {
        result, err := i.db.UnsetDevicesGroup(ctx, []model.DeviceID{id}, group)
        if err != nil {
                return errors.Wrap(err, "failed to unassign group from device")
        } else if result.MatchedCount <= 0 {
                return store.ErrDevNotFound
        }

        i.maybeTriggerReindex(ctx, []model.DeviceID{id})

        return nil
}

func (i *inventory) UpdateDevicesGroup(
        ctx context.Context,
        deviceIDs []model.DeviceID,
        group model.GroupName,
) (*model.UpdateResult, error) {

        res, err := i.db.UpdateDevicesGroup(ctx, deviceIDs, group)
        if err != nil {
                return nil, err
        }

        if i.enableReporting {
                i.triggerReindex(ctx, deviceIDs)
        }

        return res, err
}

func (i *inventory) UpdateDeviceGroup(
        ctx context.Context,
        devid model.DeviceID,
        group model.GroupName,
) error {
        result, err := i.db.UpdateDevicesGroup(
                ctx, []model.DeviceID{devid}, group,
        )
        if err != nil {
                return errors.Wrap(err, "failed to add device to group")
        } else if result.MatchedCount <= 0 {
                return store.ErrDevNotFound
        }

        i.maybeTriggerReindex(ctx, []model.DeviceID{devid})

        return nil
}

func (i *inventory) ListGroups(
        ctx context.Context,
        filters []model.FilterPredicate,
) ([]model.GroupName, error) {
        groups, err := i.db.ListGroups(ctx, filters)
        if err != nil {
                return nil, errors.Wrap(err, "failed to list groups")
        }

        if groups == nil {
                return []model.GroupName{}, nil
        }
        return groups, nil
}

func (i *inventory) ListDevicesByGroup(
        ctx context.Context,
        group model.GroupName,
        skip,
        limit int,
) ([]model.DeviceID, int, error) {
        ids, totalCount, err := i.db.GetDevicesByGroup(ctx, group, skip, limit)
        if err != nil {
                if err == store.ErrGroupNotFound {
                        return nil, -1, err
                } else {
                        return nil, -1, errors.Wrap(err, "failed to list devices by group")
                }
        }

        return ids, totalCount, nil
}

func (i *inventory) GetDeviceGroup(
        ctx context.Context,
        id model.DeviceID,
) (model.GroupName, error) {
        group, err := i.db.GetDeviceGroup(ctx, id)
        if err != nil {
                if err == store.ErrDevNotFound {
                        return "", err
                } else {
                        return "", errors.Wrap(err, "failed to get device's group")
                }
        }

        return group, nil
}

func (i *inventory) CreateTenant(ctx context.Context, tenant model.NewTenant) error {
        if err := i.db.WithAutomigrate().
                MigrateTenant(ctx, mongo.DbVersion, tenant.ID); err != nil {
                return errors.Wrapf(err, "failed to apply migrations for tenant %v", tenant.ID)
        }
        return nil
}

func (i *inventory) SearchDevices(
        ctx context.Context,
        searchParams model.SearchParams,
) ([]model.Device, int, error) {
        devs, totalCount, err := i.db.SearchDevices(ctx, searchParams)

        if err != nil {
                return nil, -1, errors.Wrap(err, "failed to fetch devices")
        }

        return devs, totalCount, nil
}

func (i *inventory) CheckAlerts(ctx context.Context, deviceId string) (int, error) {
        return i.dmClient.CheckAlerts(ctx, deviceId)
}

// maybeTriggerReindex conditionally triggers the reindex_reporting workflow for a device
func (i *inventory) maybeTriggerReindex(ctx context.Context, deviceIDs []model.DeviceID) {
        if i.enableReporting {
                i.triggerReindex(ctx, deviceIDs)
        }
}

// triggerReindex triggers the reindex_reporting workflow for a device
func (i *inventory) triggerReindex(ctx context.Context, deviceIDs []model.DeviceID) {
        err := i.wfClient.StartReindex(ctx, deviceIDs)
        if err != nil {
                l := log.FromContext(ctx)
                l.Errorf("failed to start reindex_reporting for devices %v, error: %v", deviceIDs, err)
        }
}

// reindexTextField reindex the device's text field
func (i *inventory) reindexTextField(ctx context.Context, devices []*model.Device) {
        l := log.FromContext(ctx)
        for _, device := range devices {
                text := utils.GetTextField(device)
                if device.Text != text {
                        err := i.db.UpdateDeviceText(ctx, device.ID, text)
                        if err != nil {
                                l.Errorf("failed to reindex the text field for device %v, error: %v",
                                        device.ID, err)
                        }
                }
        }
}

1	// Copyright 2023 Northern.tech AS
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	package inv
16
17	import (
18	"context"
19	"reflect"
20	"time"
21
22	"github.com/pkg/errors"
23
24	"github.com/mendersoftware/go-lib-micro/log"
25
26	"github.com/mendersoftware/inventory/client/devicemonitor"
27	"github.com/mendersoftware/inventory/client/workflows"
28	"github.com/mendersoftware/inventory/model"
29	"github.com/mendersoftware/inventory/store"
30	"github.com/mendersoftware/inventory/store/mongo"
31	"github.com/mendersoftware/inventory/utils"
32	)
33
34	const reindexBatchSize = 100
35
36	var (
37	ErrETagDoesntMatch = errors.New("ETag does not match")
38	ErrTooManyAttributes = errors.New("the number of attributes in the scope is above the limit")
39	)
40
41	// this inventory service interface
42	//
43	//go:generate ../utils/mockgen.sh
44	type InventoryApp interface {
45	WithReporting(c workflows.Client) InventoryApp
46	HealthCheck(ctx context.Context) error
47	ListDevices(ctx context.Context, q store.ListQuery) ([]model.Device, int, error)
48	GetDevice(ctx context.Context, id model.DeviceID) (*model.Device, error)
49	AddDevice(ctx context.Context, d *model.Device) error
50	UpsertAttributes(ctx context.Context, id model.DeviceID, attrs model.DeviceAttributes) error
51	UpsertAttributesWithUpdated(
52	ctx context.Context,
53	id model.DeviceID,
54	attrs model.DeviceAttributes,
55	scope string,
56	etag string,
57	) error
58	UpsertDevicesStatuses(
59	ctx context.Context,
60	devices []model.DeviceUpdate,
61	attrs model.DeviceAttributes,
62	) (*model.UpdateResult, error)
63	ReplaceAttributes(
64	ctx context.Context,
65	id model.DeviceID,
66	upsertAttrs model.DeviceAttributes,
67	scope string,
68	etag string,
69	) error
70	GetFiltersAttributes(ctx context.Context) ([]model.FilterAttribute, error)
71	DeleteGroup(ctx context.Context, groupName model.GroupName) (*model.UpdateResult, error)
72	UnsetDeviceGroup(ctx context.Context, id model.DeviceID, groupName model.GroupName) error
73	UnsetDevicesGroup(
74	ctx context.Context,
75	deviceIDs []model.DeviceID,
76	groupName model.GroupName,
77	) (*model.UpdateResult, error)
78	UpdateDeviceGroup(ctx context.Context, id model.DeviceID, group model.GroupName) error
79	UpdateDevicesGroup(
80	ctx context.Context,
81	ids []model.DeviceID,
82	group model.GroupName,
83	) (*model.UpdateResult, error)
84	ListGroups(ctx context.Context, filters []model.FilterPredicate) ([]model.GroupName, error)
85	ListDevicesByGroup(
86	ctx context.Context,
87	group model.GroupName,
88	skip int,
89	limit int,
90	) ([]model.DeviceID, int, error)
91	GetDeviceGroup(ctx context.Context, id model.DeviceID) (model.GroupName, error)
92	DeleteDevice(ctx context.Context, id model.DeviceID) error
93	DeleteDevices(
94	ctx context.Context,
95	ids []model.DeviceID,
96	) (*model.UpdateResult, error)
97	CreateTenant(ctx context.Context, tenant model.NewTenant) error
98	SearchDevices(ctx context.Context, searchParams model.SearchParams) ([]model.Device, int, error)
99	CheckAlerts(ctx context.Context, deviceId string) (int, error)
100	WithLimits(attributes, tags int) InventoryApp
101	WithDevicemonitor(client devicemonitor.Client) InventoryApp
102	}
103
104	type inventory struct {
105	db store.DataStore
106	limitAttributes int
107	limitTags int
108	dmClient devicemonitor.Client
109	enableReporting bool
110	wfClient workflows.Client
111	}
112
113	func NewInventory(d store.DataStore) InventoryApp {	7✔
114	return &inventory{db: d}	7✔
115	}	7✔
116
117	func (i *inventory) WithDevicemonitor(client devicemonitor.Client) InventoryApp {	3✔
118	i.dmClient = client	3✔
119	return i	3✔
120	}	3✔
121
122	func (i *inventory) WithLimits(limitAttributes, limitTags int) InventoryApp {	28✔
123	i.limitAttributes = limitAttributes	28✔
124	i.limitTags = limitTags	28✔
125	return i	28✔
126	}	28✔
127
128	func (i *inventory) WithReporting(client workflows.Client) InventoryApp {	13✔
129	i.enableReporting = true	13✔
130	i.wfClient = client	13✔
131	return i	13✔
132	}	13✔
133
134	func (i *inventory) HealthCheck(ctx context.Context) error {	3✔
135	err := i.db.Ping(ctx)	3✔
136	if err != nil {	4✔
137	return errors.Wrap(err, "error reaching MongoDB")	1✔
138	}	1✔
139
140	if i.enableReporting {	4✔
141	err := i.wfClient.CheckHealth(ctx)	2✔
142	if err != nil {	3✔
143	return errors.Wrap(err, "error reaching workflows")	1✔
144	}	1✔
145	}
146
147	return nil	1✔
148	}
149
150	func (i *inventory) ListDevices(
151	ctx context.Context,
152	q store.ListQuery,
153	) ([]model.Device, int, error) {	14✔
154	devs, totalCount, err := i.db.GetDevices(ctx, q)	14✔
155		14✔
156	if err != nil {	15✔
157	return nil, -1, errors.Wrap(err, "failed to fetch devices")	1✔
158	}	1✔
159
160	return devs, totalCount, nil	13✔
161	}
162
163	func (i inventory) GetDevice(ctx context.Context, id model.DeviceID) (model.Device, error) {	15✔
164	dev, err := i.db.GetDevice(ctx, id)	15✔
165	if err != nil {	16✔
166	return nil, errors.Wrap(err, "failed to fetch device")	1✔
167	}	1✔
168	return dev, nil	14✔
169	}
170
171	func (i inventory) AddDevice(ctx context.Context, dev model.Device) error {	79✔
172	if dev == nil {	80✔
173	return errors.New("no device given")	1✔
174	}	1✔
175	dev.Text = utils.GetTextField(dev)	78✔
176	err := i.db.AddDevice(ctx, dev)	78✔
177	if err != nil {	79✔
178	return errors.Wrap(err, "failed to add device")	1✔
179	}	1✔
180
181	i.maybeTriggerReindex(ctx, []model.DeviceID{dev.ID})	77✔
182		77✔
183	return nil	77✔
184	}
185
186	func (i *inventory) DeleteDevices(
187	ctx context.Context,
188	ids []model.DeviceID,
189	) (*model.UpdateResult, error) {	3✔
190	res, err := i.db.DeleteDevices(ctx, ids)	3✔
191	if err != nil {	4✔
192	return nil, err	1✔
193	}	1✔
194
195	if i.enableReporting {	4✔
196	for _, d := range ids {	4✔
197	i.triggerReindex(ctx, []model.DeviceID{d})	2✔
198	}	2✔
199	}
200
201	return res, err	2✔
202	}
203
204	func (i *inventory) DeleteDevice(ctx context.Context, id model.DeviceID) error {	3✔
205	res, err := i.db.DeleteDevices(ctx, []model.DeviceID{id})	3✔
206	if err != nil {	4✔
207	return errors.Wrap(err, "failed to delete device")	1✔
208	} else if res.DeletedCount < 1 {	4✔
209	return store.ErrDevNotFound	1✔
210	}	1✔
211	i.maybeTriggerReindex(ctx, []model.DeviceID{id})	1✔
212		1✔
213	return nil	1✔
214	}
215
216	func (i *inventory) UpsertAttributes(
217	ctx context.Context,
218	id model.DeviceID,
219	attrs model.DeviceAttributes,
220	) error {	2✔
221	res, err := i.db.UpsertDevicesAttributes(	2✔
222	ctx, []model.DeviceID{id}, attrs,	2✔
223	)	2✔
224	if err != nil {	3✔
225	return errors.Wrap(err, "failed to upsert attributes in db")	1✔
226	}	1✔
227	if res != nil && res.MatchedCount > 0 {	1✔
228	i.reindexTextField(ctx, res.Devices)	×
229	i.maybeTriggerReindex(ctx, []model.DeviceID{id})	×
230	}	×
231	return nil	1✔
232	}
233
234	func (i *inventory) checkAttributesLimits(
235	ctx context.Context,
236	id model.DeviceID,
237	attrs model.DeviceAttributes,
238	scope string,
239	) error {	19✔
240	limit := 0	19✔
241	switch scope {	19✔
242	case model.AttrScopeInventory:	7✔
243	limit = i.limitAttributes	7✔
244	case model.AttrScopeTags:	12✔
245	limit = i.limitTags	12✔
246	}
247	if limit == 0 {	25✔
248	return nil	6✔
249	}	6✔
250	device, err := i.db.GetDevice(ctx, id)	13✔
251	if err != nil && err != store.ErrDevNotFound {	14✔
252	return errors.Wrap(err, "failed to get the device")	1✔
253	} else if device == nil {	13✔
254	return nil	×
255	}	×
256	count := 0	12✔
257	for _, attr := range device.Attributes {	115✔
258	if attr.Scope == scope {	110✔
259	count += 1	7✔
260	if count > limit {	8✔
261	break	1✔
262	}
263	}
264	}
265	for _, attr := range attrs {	46✔
266	if count > limit {	36✔
267	break	2✔
268	}
269	found := false	32✔
270	for _, devAttr := range device.Attributes {	469✔
271	if attr.Scope == scope && attr.Name == devAttr.Name {	439✔
272	found = true	2✔
273	}	2✔
274	}
275	if !found {	62✔
276	count++	30✔
277	}	30✔
278	}
279	if count > limit {	17✔
280	return ErrTooManyAttributes	5✔
281	}	5✔
282	return nil	7✔
283	}
284
285	const oneDay = 24 * time.Hour
286
287	func (i *inventory) needsUpsert(
288	device *model.Device,
289	upsertAttrs model.DeviceAttributes,
290	removeAttrs model.DeviceAttributes,
291	) bool {	30✔
292	needsUpsert := true	30✔
293	if device != nil {	56✔
294	// we update the inventory attributes at least once every (calendar) day	26✔
295	if !device.UpdatedTs.IsZero() &&	26✔
296	device.UpdatedTs.Truncate(oneDay) != time.Now().Truncate(oneDay) {	27✔
297	return true	1✔
298	}	1✔
299	needsUpsert = false	25✔
300	for _, attribute := range upsertAttrs {	49✔
301	if attribute.Scope != model.AttrScopeInventory {	39✔
302	needsUpsert = true	15✔
303	break	15✔
304	}
305	attributeChanged := true	9✔
306	for _, deviceAttribute := range device.Attributes {	14✔
307	if !(attribute.Scope == deviceAttribute.Scope &&	5✔
308	attribute.Name == deviceAttribute.Name) {	7✔
309	continue	2✔
310	}
311	attributeChanged = !reflect.DeepEqual(attribute.Value, deviceAttribute.Value)	3✔
312	break	3✔
313	}
314	if attributeChanged {	16✔
315	needsUpsert = true	7✔
316	break	7✔
317	}
318	}
319	if !needsUpsert && len(removeAttrs) > 0 {	27✔
320	OuterLoop:	2✔
321	for _, attribute := range removeAttrs {	4✔
322	for _, deviceAttribute := range device.Attributes {	4✔
323	if attribute.Scope == deviceAttribute.Scope &&	2✔
324	attribute.Name == deviceAttribute.Name {	4✔
325	needsUpsert = true	2✔
326	break OuterLoop	2✔
327	}
328	}
329	}
330	}
331	}
332	return needsUpsert	29✔
333	}
334
335	func (i *inventory) UpsertAttributesWithUpdated(
336	ctx context.Context,
337	id model.DeviceID,
338	attrs model.DeviceAttributes,
339	scope string,
340	etag string,
341	) error {	19✔
342	if err := i.checkAttributesLimits(ctx, id, attrs, scope); err != nil {	25✔
343	return err	6✔
344	}	6✔
345
346	device, err := i.db.GetDevice(ctx, id)	13✔
347	if err != nil && err != store.ErrDevNotFound {	13✔
NEW 348	return errors.Wrap(err, "failed to get the device")	×
349	} else if !i.needsUpsert(device, attrs, nil) {	14✔
350	return nil	1✔
351	}	1✔
352
353	res, err := i.db.UpsertDevicesAttributesWithUpdated(	12✔
354	ctx, []model.DeviceID{id}, attrs, scope, etag,	12✔
355	)	12✔
356	if err != nil {	14✔
357	return errors.Wrap(err, "failed to upsert attributes in db")	2✔
358	}	2✔
359	if scope == model.AttrScopeTags {	17✔
360	if res != nil && res.MatchedCount == 0 && etag != "" {	8✔
361	return ErrETagDoesntMatch	1✔
362	}	1✔
363	}
364
365	if res != nil && res.MatchedCount > 0 {	18✔
366	i.reindexTextField(ctx, res.Devices)	9✔
367	i.maybeTriggerReindex(ctx, []model.DeviceID{id})	9✔
368	}	9✔
369	return nil	9✔
370	}
371
372	func getRemoveAttrs(
373	device *model.Device,
374	scope string,
375	upsertAttrs model.DeviceAttributes,
376	) model.DeviceAttributes {	17✔
377	removeAttrs := model.DeviceAttributes{}	17✔
378	if device != nil {	30✔
379	for _, attr := range device.Attributes {	121✔
380	if attr.Scope == scope {	120✔
381	update := false	12✔
382	for _, upsertAttr := range upsertAttrs {	21✔
383	if upsertAttr.Name == attr.Name {	14✔
384	update = true	5✔
385	break	5✔
386	}
387	}
388	if !update {	19✔
389	removeAttrs = append(removeAttrs, attr)	7✔
390	}	7✔
391	}
392	}
393	}
394	return removeAttrs	17✔
395	}
396
397	func (i *inventory) ReplaceAttributes(
398	ctx context.Context,
399	id model.DeviceID,
400	upsertAttrs model.DeviceAttributes,
401	scope string,
402	etag string,
403	) error {	20✔
404	limit := 0	20✔
405	switch scope {	20✔
406	case model.AttrScopeInventory:	7✔
407	limit = i.limitAttributes	7✔
408	case model.AttrScopeTags:	13✔
409	limit = i.limitTags	13✔
410	}
411	if limit > 0 && len(upsertAttrs) > limit {	22✔
412	return ErrTooManyAttributes	2✔
413	}	2✔
414
415	device, err := i.db.GetDevice(ctx, id)	18✔
416	if err != nil && err != store.ErrDevNotFound {	19✔
417	return errors.Wrap(err, "failed to get the device")	1✔
418	}	1✔
419
420	removeAttrs := getRemoveAttrs(device, scope, upsertAttrs)	17✔
421	if !i.needsUpsert(device, upsertAttrs, removeAttrs) {	17✔
NEW 422	return nil	×
UNCOV 423	}	×
424
425	res, err := i.db.UpsertRemoveDeviceAttributes(ctx, id, upsertAttrs, removeAttrs, scope, etag)	17✔
426	if err != nil {	19✔
427	return errors.Wrap(err, "failed to replace attributes in db")	2✔
428	}	2✔
429	if scope == model.AttrScopeTags {	26✔
430	if res != nil && res.MatchedCount == 0 && etag != "" {	12✔
431	return ErrETagDoesntMatch	1✔
432	}	1✔
433	}
434	if res != nil && res.MatchedCount > 0 {	27✔
435	i.reindexTextField(ctx, res.Devices)	13✔
436	i.maybeTriggerReindex(ctx, []model.DeviceID{id})	13✔
437	}	13✔
438	return nil	14✔
439	}
440
441	func (i *inventory) GetFiltersAttributes(ctx context.Context) ([]model.FilterAttribute, error) {	4✔
442	attributes, err := i.db.GetFiltersAttributes(ctx)	4✔
443	if err != nil {	5✔
444	return nil, errors.Wrap(err, "failed to get filter attributes from the db")	1✔
445	}	1✔
446	return attributes, nil	3✔
447	}
448
449	func (i *inventory) DeleteGroup(
450	ctx context.Context,
451	groupName model.GroupName,
452	) (*model.UpdateResult, error) {	3✔
453	deviceIDs, err := i.db.DeleteGroup(ctx, groupName)	3✔
454	if err != nil {	4✔
455	return nil, errors.Wrap(err, "failed to delete group")	1✔
456	}	1✔
457
458	batchDeviceIDsLength := 0	2✔
459	batchDeviceIDs := make([]model.DeviceID, reindexBatchSize)	2✔
460		2✔
461	triggerReindex := func() {	5✔
462	i.maybeTriggerReindex(ctx, batchDeviceIDs[0:batchDeviceIDsLength])	3✔
463	batchDeviceIDsLength = 0	3✔
464	}	3✔
465
466	res := &model.UpdateResult{}	2✔
467	for deviceID := range deviceIDs {	106✔
468	batchDeviceIDs[batchDeviceIDsLength] = deviceID	104✔
469	batchDeviceIDsLength++	104✔
470	if batchDeviceIDsLength == reindexBatchSize {	105✔
471	triggerReindex()	1✔
472	}	1✔
473	res.MatchedCount += 1	104✔
474	res.UpdatedCount += 1	104✔
475	}
476	if batchDeviceIDsLength > 0 {	4✔
477	triggerReindex()	2✔
478	}	2✔
479
480	return res, err	2✔
481	}
482
483	func (i *inventory) UpsertDevicesStatuses(
484	ctx context.Context,
485	devices []model.DeviceUpdate,
486	attrs model.DeviceAttributes,
487	) (*model.UpdateResult, error) {	3✔
488	res, err := i.db.UpsertDevicesAttributesWithRevision(ctx, devices, attrs)	3✔
489	if err != nil {	4✔
490	return nil, err	1✔
491	}	1✔
492
493	if i.enableReporting {	4✔
494	deviceIDs := make([]model.DeviceID, len(devices))	2✔
495	for i, d := range devices {	4✔
496	deviceIDs[i] = d.Id	2✔
497	}	2✔
498	i.triggerReindex(ctx, deviceIDs)	2✔
499	}
500
501	return res, err	2✔
502	}
503
504	func (i *inventory) UnsetDevicesGroup(
505	ctx context.Context,
506	deviceIDs []model.DeviceID,
507	groupName model.GroupName,
508	) (*model.UpdateResult, error) {	2✔
509	res, err := i.db.UnsetDevicesGroup(ctx, deviceIDs, groupName)	2✔
510	if err != nil {	3✔
511	return nil, err	1✔
512	}	1✔
513
514	if i.enableReporting {	1✔
515	i.triggerReindex(ctx, deviceIDs)	×
516	}	×
517
518	return res, nil	1✔
519	}
520
521	func (i *inventory) UnsetDeviceGroup(
522	ctx context.Context,
523	id model.DeviceID,
524	group model.GroupName,
525	) error {	8✔
526	result, err := i.db.UnsetDevicesGroup(ctx, []model.DeviceID{id}, group)	8✔
527	if err != nil {	9✔
528	return errors.Wrap(err, "failed to unassign group from device")	1✔
529	} else if result.MatchedCount <= 0 {	12✔
530	return store.ErrDevNotFound	4✔
531	}	4✔
532
533	i.maybeTriggerReindex(ctx, []model.DeviceID{id})	3✔
534		3✔
535	return nil	3✔
536	}
537
538	func (i *inventory) UpdateDevicesGroup(
539	ctx context.Context,
540	deviceIDs []model.DeviceID,
541	group model.GroupName,
542	) (*model.UpdateResult, error) {	2✔
543		2✔
544	res, err := i.db.UpdateDevicesGroup(ctx, deviceIDs, group)	2✔
545	if err != nil {	3✔
546	return nil, err	1✔
547	}	1✔
548
549	if i.enableReporting {	1✔
550	i.triggerReindex(ctx, deviceIDs)	×
551	}	×
552
553	return res, err	1✔
554	}
555
556	func (i *inventory) UpdateDeviceGroup(
557	ctx context.Context,
558	devid model.DeviceID,
559	group model.GroupName,
560	) error {	52✔
561	result, err := i.db.UpdateDevicesGroup(	52✔
562	ctx, []model.DeviceID{devid}, group,	52✔
563	)	52✔
564	if err != nil {	52✔
565	return errors.Wrap(err, "failed to add device to group")	×
566	} else if result.MatchedCount <= 0 {	53✔
567	return store.ErrDevNotFound	1✔
568	}	1✔
569
570	i.maybeTriggerReindex(ctx, []model.DeviceID{devid})	51✔
571		51✔
572	return nil	51✔
573	}
574
575	func (i *inventory) ListGroups(
576	ctx context.Context,
577	filters []model.FilterPredicate,
578	) ([]model.GroupName, error) {	10✔
579	groups, err := i.db.ListGroups(ctx, filters)	10✔
580	if err != nil {	11✔
581	return nil, errors.Wrap(err, "failed to list groups")	1✔
582	}	1✔
583
584	if groups == nil {	10✔
585	return []model.GroupName{}, nil	1✔
586	}	1✔
587	return groups, nil	8✔
588	}
589
590	func (i *inventory) ListDevicesByGroup(
591	ctx context.Context,
592	group model.GroupName,
593	skip,
594	limit int,
595	) ([]model.DeviceID, int, error) {	26✔
596	ids, totalCount, err := i.db.GetDevicesByGroup(ctx, group, skip, limit)	26✔
597	if err != nil {	32✔
598	if err == store.ErrGroupNotFound {	11✔
599	return nil, -1, err	5✔
600	} else {	6✔
601	return nil, -1, errors.Wrap(err, "failed to list devices by group")	1✔
602	}	1✔
603	}
604
605	return ids, totalCount, nil	20✔
606	}
607
608	func (i *inventory) GetDeviceGroup(
609	ctx context.Context,
610	id model.DeviceID,
611	) (model.GroupName, error) {	4✔
612	group, err := i.db.GetDeviceGroup(ctx, id)	4✔
613	if err != nil {	6✔
614	if err == store.ErrDevNotFound {	3✔
615	return "", err	1✔
616	} else {	2✔
617	return "", errors.Wrap(err, "failed to get device's group")	1✔
618	}	1✔
619	}
620
621	return group, nil	2✔
622	}
623
624	func (i *inventory) CreateTenant(ctx context.Context, tenant model.NewTenant) error {	5✔
625	if err := i.db.WithAutomigrate().	5✔
626	MigrateTenant(ctx, mongo.DbVersion, tenant.ID); err != nil {	6✔
627	return errors.Wrapf(err, "failed to apply migrations for tenant %v", tenant.ID)	1✔
628	}	1✔
629	return nil	4✔
630	}
631
632	func (i *inventory) SearchDevices(
633	ctx context.Context,
634	searchParams model.SearchParams,
635	) ([]model.Device, int, error) {	2✔
636	devs, totalCount, err := i.db.SearchDevices(ctx, searchParams)	2✔
637		2✔
638	if err != nil {	3✔
639	return nil, -1, errors.Wrap(err, "failed to fetch devices")	1✔
640	}	1✔
641
642	return devs, totalCount, nil	1✔
643	}
644
645	func (i *inventory) CheckAlerts(ctx context.Context, deviceId string) (int, error) {	2✔
646	return i.dmClient.CheckAlerts(ctx, deviceId)	2✔
647	}	2✔
648
649	// maybeTriggerReindex conditionally triggers the reindex_reporting workflow for a device
650	func (i *inventory) maybeTriggerReindex(ctx context.Context, deviceIDs []model.DeviceID) {	157✔
651	if i.enableReporting {	297✔
652	i.triggerReindex(ctx, deviceIDs)	140✔
653	}	140✔
654	}
655
656	// triggerReindex triggers the reindex_reporting workflow for a device
657	func (i *inventory) triggerReindex(ctx context.Context, deviceIDs []model.DeviceID) {	144✔
658	err := i.wfClient.StartReindex(ctx, deviceIDs)	144✔
659	if err != nil {	283✔
660	l := log.FromContext(ctx)	139✔
661	l.Errorf("failed to start reindex_reporting for devices %v, error: %v", deviceIDs, err)	139✔
662	}	139✔
663	}
664
665	// reindexTextField reindex the device's text field
666	func (i inventory) reindexTextField(ctx context.Context, devices []model.Device) {	22✔
667	l := log.FromContext(ctx)	22✔
668	for _, device := range devices {	44✔
669	text := utils.GetTextField(device)	22✔
670	if device.Text != text {	42✔
671	err := i.db.UpdateDeviceText(ctx, device.ID, text)	20✔
672	if err != nil {	20✔
673	l.Errorf("failed to reindex the text field for device %v, error: %v",	×
674	device.ID, err)	×
675	}	×
676	}
677	}
678	}

mendersoftware / inventory / 989241608

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous