955593424

pending completion

Build # 955593424

Build Type

Pull #401

gitlab-ci

Committed by

merlin-northern

Commit Message

fix: attributes udpate: be mindful of the subset of attributes being patched.

Changelog: Title
Ticket: MEN-6643
Signed-off-by: Peter Grzybowski <peter@northern.tech>

Pull Request Pull Request #401: fix: attributes udpate: be mindful of the subset of attributes being …

Run Details

15 of 38 new or added lines in 3 files covered. (39.47%)

21 existing lines in 2 files now uncovered.

3200 of 3562 relevant lines covered (89.84%)

135.41 hits per line

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

90.99

/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"
        "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
        WithLastUpdateDurationThreshold(threshold time.Duration) InventoryApp
        InventoryNeedsUpdate(
                ctx context.Context,
                newAttributes model.DeviceAttributes,
                deviceID model.DeviceID,
                scope string,
        ) bool
}

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

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) WithLastUpdateDurationThreshold(threshold time.Duration) InventoryApp {
        i.lastUpdateDurationThreshold = threshold
        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
}

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
        }
        res, err := i.db.UpsertDevicesAttributesWithUpdated(
                ctx, []model.DeviceID{id}, attrs, scope, etag, i.lastUpdateDurationThreshold,
        )
        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 (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 := 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)
                                }
                        }
                }
        }

        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)
                        }
                }
        }
}

// InventoryNeedsUpdate returns true if given attributes need an update
// The need is determined as follows:
//  1. if scope is not inventory we assume we need to update
//  2. if the new attributes differ from currently existing in the db we need to update
//     a. we make both existing and new attributes into a map with keys of scope_name
//     b. if all the new attributes exits in the db we consider the update as not needed
//     c. if any of the new attributes does not exist or has different value in the db
//     we assume we have to update
func (i *inventory) InventoryNeedsUpdate(
        ctx context.Context,
        newAttributes model.DeviceAttributes,
        deviceID model.DeviceID,
        scope string,
) bool {
        if scope != model.AttrScopeInventory {
                // we always update non-inventory scope
                return true
        }

        newAttributesMap := newAttributes.ToMap(model.AttrNameUpdated, model.AttrNameCreated)
        existingDevice, err := i.db.GetDevice(ctx, deviceID)
        if err != nil || existingDevice == nil {
                return true
        }
        existingAttributes := existingDevice.Attributes.ToMap(
                model.AttrNameUpdated,
                model.AttrNameCreated,
        )
        for k, v := range newAttributesMap {
                if e, ok := existingAttributes[k]; ok {
                        if !v.Equal(e) {
                                return true
                        }
                } else {
                        // cannot be equal if keys differ (the last possibility)
                        return true
                }
        }

        return false
}

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

mendersoftware / inventory / 955593424

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