1565407895

Committed 29 Nov 2024 07:58AM UTC coverage: 91.502%. Remained the same

Build # 1565407895

Build Type

push

gitlab-ci

Committed by

web-flow

Commit Message

Merge pull request #464 from alfrunes/4.4.x

chore(deps): Upgrade docker image to golang:1.23.3-alpine3.20

Run Details

3144 of 3436 relevant lines covered (91.5%)

151.63 hits per line

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

95.74

/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/go-lib-micro/log"

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

const reindexBatchSize = 100

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

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

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

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

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

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

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

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

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

        return nil
}

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

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

        return devs, totalCount, nil
}

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

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

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

        return nil
}

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

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

        return res, err
}

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

        return nil
}

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

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

const oneDay = 24 * time.Hour

func (i *inventory) needsUpsert(
        device *model.Device,
        upsertAttrs model.DeviceAttributes,
        removeAttrs model.DeviceAttributes,
) bool {
        needsUpsert := true
        if device != nil {
                // we update the inventory attributes at least once every (calendar) day
                if !device.UpdatedTs.IsZero() &&
                        device.UpdatedTs.Truncate(oneDay) != time.Now().Truncate(oneDay) {
                        return true
                }
                needsUpsert = false
                for _, attribute := range upsertAttrs {
                        if attribute.Scope != model.AttrScopeInventory {
                                needsUpsert = true
                                break
                        }
                        attributeChanged := true
                        for _, deviceAttribute := range device.Attributes {
                                if !(attribute.Scope == deviceAttribute.Scope &&
                                        attribute.Name == deviceAttribute.Name) {
                                        continue
                                }
                                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/go-lib-micro/log"
26
27	"github.com/mendersoftware/inventory/client/devicemonitor"
28	"github.com/mendersoftware/inventory/client/workflows"
29	"github.com/mendersoftware/inventory/model"
30	"github.com/mendersoftware/inventory/store"
31	"github.com/mendersoftware/inventory/store/mongo"
32	"github.com/mendersoftware/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 {	7✔
115	return &inventory{db: d}	7✔
116	}	7✔
117
118	func (i *inventory) WithDevicemonitor(client devicemonitor.Client) InventoryApp {	3✔
119	i.dmClient = client	3✔
120	return i	3✔
121	}	3✔
122
123	func (i *inventory) WithLimits(limitAttributes, limitTags int) InventoryApp {	29✔
124	i.limitAttributes = limitAttributes	29✔
125	i.limitTags = limitTags	29✔
126	return i	29✔
127	}	29✔
128
129	func (i *inventory) WithReporting(client workflows.Client) InventoryApp {	12✔
130	i.enableReporting = true	12✔
131	i.wfClient = client	12✔
132	return i	12✔
133	}	12✔
134
135	func (i *inventory) HealthCheck(ctx context.Context) error {	3✔
136	err := i.db.Ping(ctx)	3✔
137	if err != nil {	4✔
138	return errors.Wrap(err, "error reaching MongoDB")	1✔
139	}	1✔
140
141	if i.enableReporting {	4✔
142	err := i.wfClient.CheckHealth(ctx)	2✔
143	if err != nil {	3✔
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) {	14✔
155	devs, totalCount, err := i.db.GetDevices(ctx, q)	14✔
156		14✔
157	if err != nil {	15✔
158	return nil, -1, errors.Wrap(err, "failed to fetch devices")	1✔
159	}	1✔
160
161	return devs, totalCount, nil	13✔
162	}
163
164	func (i inventory) GetDevice(ctx context.Context, id model.DeviceID) (model.Device, error) {	15✔
165	dev, err := i.db.GetDevice(ctx, id)	15✔
166	if err != nil {	16✔
167	return nil, errors.Wrap(err, "failed to fetch device")	1✔
168	}	1✔
169	return dev, nil	14✔
170	}
171
172	func (i inventory) AddDevice(ctx context.Context, dev model.Device) error {	79✔
173	if dev == nil {	80✔
174	return errors.New("no device given")	1✔
175	}	1✔
176	dev.Text = utils.GetTextField(dev)	78✔
177	err := i.db.AddDevice(ctx, dev)	78✔
178	if err != nil {	79✔
179	return errors.Wrap(err, "failed to add device")	1✔
180	}	1✔
181
182	i.maybeTriggerReindex(ctx, []model.DeviceID{dev.ID})	77✔
183		77✔
184	return nil	77✔
185	}
186
187	func (i *inventory) DeleteDevices(
188	ctx context.Context,
189	ids []model.DeviceID,
190	) (*model.UpdateResult, error) {	3✔
191	res, err := i.db.DeleteDevices(ctx, ids)	3✔
192	if err != nil {	4✔
193	return nil, err	1✔
194	}	1✔
195
196	if i.enableReporting {	4✔
197	for _, d := range ids {	4✔
198	i.triggerReindex(ctx, []model.DeviceID{d})	2✔
199	}	2✔
200	}
201
202	return res, err	2✔
203	}
204
205	func (i *inventory) DeleteDevice(ctx context.Context, id model.DeviceID) error {	3✔
206	res, err := i.db.DeleteDevices(ctx, []model.DeviceID{id})	3✔
207	if err != nil {	4✔
208	return errors.Wrap(err, "failed to delete device")	1✔
209	} else if res.DeletedCount < 1 {	4✔
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 {	2✔
222	res, err := i.db.UpsertDevicesAttributes(	2✔
223	ctx, []model.DeviceID{id}, attrs,	2✔
224	)	2✔
225	if err != nil {	3✔
226	return errors.Wrap(err, "failed to upsert attributes in db")	1✔
227	}	1✔
228	if res != nil && res.MatchedCount > 0 {	1✔
229	i.reindexTextField(ctx, res.Devices)	×
230	i.maybeTriggerReindex(ctx, []model.DeviceID{id})	×
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 {	20✔
241	limit := 0	20✔
242	switch scope {	20✔
243	case model.AttrScopeInventory:	8✔
244	limit = i.limitAttributes	8✔
245	case model.AttrScopeTags:	12✔
246	limit = i.limitTags	12✔
247	}
248	if limit == 0 {	27✔
249	return nil	7✔
250	}	7✔
251	device, err := i.db.GetDevice(ctx, id)	13✔
252	if err != nil && err != store.ErrDevNotFound {	14✔
253	return errors.Wrap(err, "failed to get the device")	1✔
254	} else if device == nil {	13✔
255	return nil	×
256	}	×
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 {	46✔
267	if count > limit {	36✔
268	break	2✔
269	}
270	found := false	32✔
271	for _, devAttr := range device.Attributes {	469✔
272	if attr.Scope == scope && attr.Name == devAttr.Name {	439✔
273	found = true	2✔
274	}	2✔
275	}
276	if !found {	62✔
277	count++	30✔
278	}	30✔
279	}
280	if count > limit {	17✔
281	return ErrTooManyAttributes	5✔
282	}	5✔
283	return nil	7✔
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 {	31✔
293	needsUpsert := true	31✔
294	if device != nil {	58✔
295	// we update the inventory attributes at least once every (calendar) day	27✔
296	if !device.UpdatedTs.IsZero() &&	27✔
297	device.UpdatedTs.Truncate(oneDay) != time.Now().Truncate(oneDay) {	28✔
298	return true	1✔
299	}	1✔
300	needsUpsert = false	26✔
301	for _, attribute := range upsertAttrs {	51✔
302	if attribute.Scope != model.AttrScopeInventory {	40✔
303	needsUpsert = true	15✔
304	break	15✔
305	}
306	attributeChanged := true	10✔
307	for _, deviceAttribute := range device.Attributes {	16✔
308	if !(attribute.Scope == deviceAttribute.Scope &&	6✔
309	attribute.Name == deviceAttribute.Name) {	8✔
310	continue	2✔
311	}
312	if _, ok := deviceAttribute.Value.(primitive.A); ok {	5✔
313	attributeChanged = !reflect.DeepEqual(attribute.Value,	1✔
314	[]interface{}(deviceAttribute.Value.(primitive.A)))	1✔
315	} else {	4✔
316	attributeChanged = !reflect.DeepEqual(attribute.Value, deviceAttribute.Value)	3✔
317	}	3✔
318	break	4✔
319	}
320	if attributeChanged {	17✔
321	needsUpsert = true	7✔
322	break	7✔
323	}
324	}
325	if !needsUpsert && len(removeAttrs) > 0 {	28✔
326	OuterLoop:	2✔
327	for _, attribute := range removeAttrs {	4✔
328	for _, deviceAttribute := range device.Attributes {	4✔
329	if attribute.Scope == deviceAttribute.Scope &&	2✔
330	attribute.Name == deviceAttribute.Name {	4✔
331	needsUpsert = true	2✔
332	break OuterLoop	2✔
333	}
334	}
335	}
336	}
337	}
338	return needsUpsert	30✔
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 {	20✔
348	if err := i.checkAttributesLimits(ctx, id, attrs, scope); err != nil {	26✔
349	return err	6✔
350	}	6✔
351
352	device, err := i.db.GetDevice(ctx, id)	14✔
353	if err != nil && err != store.ErrDevNotFound {	14✔
354	return errors.Wrap(err, "failed to get the device")	×
355	} else if !i.needsUpsert(device, attrs, nil) {	16✔
356	return nil	2✔
357	}	2✔
358
359	res, err := i.db.UpsertDevicesAttributesWithUpdated(	12✔
360	ctx, []model.DeviceID{id}, attrs, scope, etag,	12✔
361	)	12✔
362	if err != nil {	14✔
363	return errors.Wrap(err, "failed to upsert attributes in db")	2✔
364	}	2✔
365	if scope == model.AttrScopeTags {	17✔
366	if res != nil && res.MatchedCount == 0 && etag != "" {	8✔
367	return ErrETagDoesntMatch	1✔
368	}	1✔
369	}
370
371	if res != nil && res.MatchedCount > 0 {	18✔
372	i.reindexTextField(ctx, res.Devices)	9✔
373	i.maybeTriggerReindex(ctx, []model.DeviceID{id})	9✔
374	}	9✔
375	return nil	9✔
376	}
377
378	func getRemoveAttrs(
379	device *model.Device,
380	scope string,
381	upsertAttrs model.DeviceAttributes,
382	) model.DeviceAttributes {	17✔
383	removeAttrs := model.DeviceAttributes{}	17✔
384	if device != nil {	30✔
385	for _, attr := range device.Attributes {	121✔
386	if attr.Scope == scope {	120✔
387	update := false	12✔
388	for _, upsertAttr := range upsertAttrs {	21✔
389	if upsertAttr.Name == attr.Name {	14✔
390	update = true	5✔
391	break	5✔
392	}
393	}
394	if !update {	19✔
395	removeAttrs = append(removeAttrs, attr)	7✔
396	}	7✔
397	}
398	}
399	}
400	return removeAttrs	17✔
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 {	20✔
410	limit := 0	20✔
411	switch scope {	20✔
412	case model.AttrScopeInventory:	7✔
413	limit = i.limitAttributes	7✔
414	case model.AttrScopeTags:	13✔
415	limit = i.limitTags	13✔
416	}
417	if limit > 0 && len(upsertAttrs) > limit {	22✔
418	return ErrTooManyAttributes	2✔
419	}	2✔
420
421	device, err := i.db.GetDevice(ctx, id)	18✔
422	if err != nil && err != store.ErrDevNotFound {	19✔
423	return errors.Wrap(err, "failed to get the device")	1✔
424	}	1✔
425
426	removeAttrs := getRemoveAttrs(device, scope, upsertAttrs)	17✔
427	if !i.needsUpsert(device, upsertAttrs, removeAttrs) {	17✔
428	return nil	×
429	}	×
430
431	res, err := i.db.UpsertRemoveDeviceAttributes(ctx, id, upsertAttrs, removeAttrs, scope, etag)	17✔
432	if err != nil {	19✔
433	return errors.Wrap(err, "failed to replace attributes in db")	2✔
434	}	2✔
435	if scope == model.AttrScopeTags {	26✔
436	if res != nil && res.MatchedCount == 0 && etag != "" {	12✔
437	return ErrETagDoesntMatch	1✔
438	}	1✔
439	}
440	if res != nil && res.MatchedCount > 0 {	27✔
441	i.reindexTextField(ctx, res.Devices)	13✔
442	i.maybeTriggerReindex(ctx, []model.DeviceID{id})	13✔
443	}	13✔
444	return nil	14✔
445	}
446
447	func (i *inventory) GetFiltersAttributes(ctx context.Context) ([]model.FilterAttribute, error) {	4✔
448	attributes, err := i.db.GetFiltersAttributes(ctx)	4✔
449	if err != nil {	5✔
450	return nil, errors.Wrap(err, "failed to get filter attributes from the db")	1✔
451	}	1✔
452	return attributes, nil	3✔
453	}
454
455	func (i *inventory) DeleteGroup(
456	ctx context.Context,
457	groupName model.GroupName,
458	) (*model.UpdateResult, error) {	3✔
459	deviceIDs, err := i.db.DeleteGroup(ctx, groupName)	3✔
460	if err != nil {	4✔
461	return nil, errors.Wrap(err, "failed to delete group")	1✔
462	}	1✔
463
464	batchDeviceIDsLength := 0	2✔
465	batchDeviceIDs := make([]model.DeviceID, reindexBatchSize)	2✔
466		2✔
467	triggerReindex := func() {	5✔
468	i.maybeTriggerReindex(ctx, batchDeviceIDs[0:batchDeviceIDsLength])	3✔
469	batchDeviceIDsLength = 0	3✔
470	}	3✔
471
472	res := &model.UpdateResult{}	2✔
473	for deviceID := range deviceIDs {	106✔
474	batchDeviceIDs[batchDeviceIDsLength] = deviceID	104✔
475	batchDeviceIDsLength++	104✔
476	if batchDeviceIDsLength == reindexBatchSize {	105✔
477	triggerReindex()	1✔
478	}	1✔
479	res.MatchedCount += 1	104✔
480	res.UpdatedCount += 1	104✔
481	}
482	if batchDeviceIDsLength > 0 {	4✔
483	triggerReindex()	2✔
484	}	2✔
485
486	return res, err	2✔
487	}
488
489	func (i *inventory) UpsertDevicesStatuses(
490	ctx context.Context,
491	devices []model.DeviceUpdate,
492	attrs model.DeviceAttributes,
493	) (*model.UpdateResult, error) {	3✔
494	res, err := i.db.UpsertDevicesAttributesWithRevision(ctx, devices, attrs)	3✔
495	if err != nil {	4✔
496	return nil, err	1✔
497	}	1✔
498
499	if i.enableReporting {	4✔
500	deviceIDs := make([]model.DeviceID, len(devices))	2✔
501	for i, d := range devices {	4✔
502	deviceIDs[i] = d.Id	2✔
503	}	2✔
504	i.triggerReindex(ctx, deviceIDs)	2✔
505	}
506
507	return res, err	2✔
508	}
509
510	func (i *inventory) UnsetDevicesGroup(
511	ctx context.Context,
512	deviceIDs []model.DeviceID,
513	groupName model.GroupName,
514	) (*model.UpdateResult, error) {	2✔
515	res, err := i.db.UnsetDevicesGroup(ctx, deviceIDs, groupName)	2✔
516	if err != nil {	3✔
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 {	8✔
532	result, err := i.db.UnsetDevicesGroup(ctx, []model.DeviceID{id}, group)	8✔
533	if err != nil {	9✔
534	return errors.Wrap(err, "failed to unassign group from device")	1✔
535	} else if result.MatchedCount <= 0 {	12✔
536	return store.ErrDevNotFound	4✔
537	}	4✔
538
539	i.maybeTriggerReindex(ctx, []model.DeviceID{id})	3✔
540		3✔
541	return nil	3✔
542	}
543
544	func (i *inventory) UpdateDevicesGroup(
545	ctx context.Context,
546	deviceIDs []model.DeviceID,
547	group model.GroupName,
548	) (*model.UpdateResult, error) {	2✔
549		2✔
550	res, err := i.db.UpdateDevicesGroup(ctx, deviceIDs, group)	2✔
551	if err != nil {	3✔
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 {	52✔
567	result, err := i.db.UpdateDevicesGroup(	52✔
568	ctx, []model.DeviceID{devid}, group,	52✔
569	)	52✔
570	if err != nil {	52✔
571	return errors.Wrap(err, "failed to add device to group")	×
572	} else if result.MatchedCount <= 0 {	53✔
573	return store.ErrDevNotFound	1✔
574	}	1✔
575
576	i.maybeTriggerReindex(ctx, []model.DeviceID{devid})	51✔
577		51✔
578	return nil	51✔
579	}
580
581	func (i *inventory) ListGroups(
582	ctx context.Context,
583	filters []model.FilterPredicate,
584	) ([]model.GroupName, error) {	10✔
585	groups, err := i.db.ListGroups(ctx, filters)	10✔
586	if err != nil {	11✔
587	return nil, errors.Wrap(err, "failed to list groups")	1✔
588	}	1✔
589
590	if groups == nil {	10✔
591	return []model.GroupName{}, nil	1✔
592	}	1✔
593	return groups, nil	8✔
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) {	26✔
602	ids, totalCount, err := i.db.GetDevicesByGroup(ctx, group, skip, limit)	26✔
603	if err != nil {	32✔
604	if err == store.ErrGroupNotFound {	11✔
605	return nil, -1, err	5✔
606	} else {	6✔
607	return nil, -1, errors.Wrap(err, "failed to list devices by group")	1✔
608	}	1✔
609	}
610
611	return ids, totalCount, nil	20✔
612	}
613
614	func (i *inventory) GetDeviceGroup(
615	ctx context.Context,
616	id model.DeviceID,
617	) (model.GroupName, error) {	4✔
618	group, err := i.db.GetDeviceGroup(ctx, id)	4✔
619	if err != nil {	6✔
620	if err == store.ErrDevNotFound {	3✔
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	2✔
628	}
629
630	func (i *inventory) CreateTenant(ctx context.Context, tenant model.NewTenant) error {	5✔
631	if err := i.db.WithAutomigrate().	5✔
632	MigrateTenant(ctx, mongo.DbVersion, tenant.ID); err != nil {	6✔
633	return errors.Wrapf(err, "failed to apply migrations for tenant %v", tenant.ID)	1✔
634	}	1✔
635	return nil	4✔
636	}
637
638	func (i *inventory) SearchDevices(
639	ctx context.Context,
640	searchParams model.SearchParams,
641	) ([]model.Device, int, error) {	2✔
642	devs, totalCount, err := i.db.SearchDevices(ctx, searchParams)	2✔
643		2✔
644	if err != nil {	3✔
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) {	2✔
652	return i.dmClient.CheckAlerts(ctx, deviceId)	2✔
653	}	2✔
654
655	// maybeTriggerReindex conditionally triggers the reindex_reporting workflow for a device
656	func (i *inventory) maybeTriggerReindex(ctx context.Context, deviceIDs []model.DeviceID) {	157✔
657	if i.enableReporting {	160✔
658	i.triggerReindex(ctx, deviceIDs)	3✔
659	}	3✔
660	}
661
662	// triggerReindex triggers the reindex_reporting workflow for a device
663	func (i *inventory) triggerReindex(ctx context.Context, deviceIDs []model.DeviceID) {	7✔
664	err := i.wfClient.StartReindex(ctx, deviceIDs)	7✔
665	if err != nil {	9✔
666	l := log.FromContext(ctx)	2✔
667	l.Errorf("failed to start reindex_reporting for devices %v, error: %v", deviceIDs, err)	2✔
668	}	2✔
669	}
670
671	// reindexTextField reindex the device's text field
672	func (i inventory) reindexTextField(ctx context.Context, devices []model.Device) {	22✔
673	l := log.FromContext(ctx)	22✔
674	for _, device := range devices {	44✔
675	text := utils.GetTextField(device)	22✔
676	if device.Text != text {	42✔
677	err := i.db.UpdateDeviceText(ctx, device.ID, text)	20✔
678	if err != nil {	20✔
679	l.Errorf("failed to reindex the text field for device %v, error: %v",	×
680	device.ID, err)	×
681	}	×
682	}
683	}
684	}

mendersoftware / inventory / 1565407895

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