959331648

pending completion

Build # 959331648

Build Type

Pull #404

gitlab-ci

Committed by

alfrunes

Commit Message

ci(qa): Update license header

Signed-off-by: Alf-Rune Siqveland <alf.rune@northern.tech>

Pull Request Pull Request #404: MEN-6425: Only update device inventory when changed

Run Details

141 of 145 new or added lines in 4 files covered. (97.24%)

6 existing lines in 2 files now uncovered.

3190 of 3482 relevant lines covered (91.61%)

147.24 hits per line

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

95.09

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

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

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

var _ InventoryApp = &inventory{}

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

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

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

func (i *inventory) WithForceUpdate(after time.Duration) *inventory {
        i.updateAfter = after
        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,
        device *model.Device,
        attrs model.DeviceAttributes,
        scope string,
) error {
        if device == nil {
                return nil
        }
        limit := 0
        switch scope {
        case model.AttrScopeInventory:
                limit = i.limitAttributes
        case model.AttrScopeTags:
                limit = i.limitTags
        }
        if limit == 0 {
                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
}

// computeAttributesUpdate computes the result of applying update on device.Attributes
// and reports whether an update must be performed. If removeAttr is present
// it is called for every attribute that should remove on replace semantics.
func (app *inventory) computeAttributesUpdate(
        device *model.Device,
        update *model.DeviceAttributes,
        scope string,
        removeAttr func(model.DeviceAttribute),
) (attrsChanged bool) {
        if update == nil {
                return false
        }
        update.MaybeInitialize()
        updateAttrs := *update
        var deviceAttrs model.DeviceAttributes
        if device != nil {
                deviceAttrs = device.Attributes.InScope(scope)
                deviceAttrs.MaybeInitialize()
        }
        i, j := 0, 0
        for i < len(updateAttrs) && j < len(deviceAttrs) {
                if updateAttrs[i].Name < deviceAttrs[j].Name {
                        // Added attribute
                        attrsChanged = true
                        i++
                } else if updateAttrs[i].Name > deviceAttrs[j].Name {
                        // deviceAttrs[j] not in updateAttrs
                        if removeAttr != nil {
                                removeAttr(deviceAttrs[j])
                                attrsChanged = true
                        }
                        j++
                } else {
                        // Same scope and name: compare value
                        if !reflect.DeepEqual(updateAttrs[i].Value, deviceAttrs[j].Value) {
                                attrsChanged = true
                        }
                        i++
                        j++
                }
        }

        // Complete the loop for i and j
        if i < len(updateAttrs) {
                attrsChanged = true
        } else if j < len(deviceAttrs) {
                // Only replace cares about dangling items
                if removeAttr != nil {
                        attrsChanged = true
                        // deviceAttrs[j] not in update
                        for j < len(deviceAttrs) {
                                removeAttr(deviceAttrs[j])
                                j++
                        }
                }
        }
        if app.updateAfter > 0 {
                // Forced update
                attrsChanged = attrsChanged ||
                        time.Since(device.UpdatedTs) > app.updateAfter
        }

        return attrsChanged
}

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

mendersoftware / inventory / 959331648

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