988981077

Committed 01 Sep 2023 10:20AM UTC coverage: 91.608% (+0.6%) from 91.042%

Build # 988981077

Build Type

Pull #409

gitlab-ci

Committed by

tranchitella

Commit Message

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

Changelog: Title
Ticket: MEN-6643

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

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

Run Details

54 of 56 new or added lines in 2 files covered. (96.43%)

1 existing line in 1 file now uncovered.

3133 of 3420 relevant lines covered (91.61%)

140.67 hits per line

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

95.66

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

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

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

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

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

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

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

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

        return nil
}

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

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

        return devs, totalCount, nil
}

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

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

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

        return nil
}

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

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

        return res, err
}

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

        return nil
}

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

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

const oneDay = 24 * time.Hour

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

mendersoftware / inventory / 988981077

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