1622978334

Committed 13 Jan 2025 03:51PM UTC coverage: 72.802% (-3.8%) from 76.608%

Build # 1622978334

Build Type

Pull #300

gitlab-ci

Committed by

alfrunes

Commit Message

fix: Deployment device count should not exceed max devices

Added a condition to skip deployments when the device count reaches max
devices.

Changelog: Title
Ticket: MEN-7847
Signed-off-by: Alf-Rune Siqveland <alf.rune@northern.tech>

Pull Request Pull Request #300: fix: Deployment device count should not exceed max devices

Run Details

4251 of 6164 branches covered (68.96%)

Branch coverage included in aggregate %.

0 of 18 new or added lines in 1 file covered. (0.0%)

2544 existing lines in 83 files now uncovered.

42741 of 58384 relevant lines covered (73.21%)

21.49 hits per line

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

94.68

/backend/services/inventory/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"
        "go.mongodb.org/mongo-driver/bson/primitive"

        "github.com/mendersoftware/mender-server/pkg/log"

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

mendersoftware / mender-server / 1622978334

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