1284672998

Committed 09 May 2024 01:19PM UTC coverage: 81.658% (-1.1%) from 82.796%

Build # 1284672998

Build Type

Pull #715

gitlab-ci

Committed by

alfrunes

Commit Message

test(acceptance/os): :broom: Remove unused fixtures

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

Pull Request Pull Request #715: Acceptance test fixup

Run Details

4808 of 5888 relevant lines covered (81.66%)

51.13 hits per line

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

67.41

/cmd/commands.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 cmd

import (
        "context"
        "fmt"
        "time"

        "github.com/mendersoftware/go-lib-micro/config"
        "github.com/mendersoftware/go-lib-micro/identity"
        "github.com/mendersoftware/go-lib-micro/log"
        "github.com/mendersoftware/go-lib-micro/mongo/migrate"
        mstore "github.com/mendersoftware/go-lib-micro/store"
        "github.com/pkg/errors"

        cinv "github.com/mendersoftware/deviceauth/client/inventory"
        "github.com/mendersoftware/deviceauth/client/orchestrator"
        "github.com/mendersoftware/deviceauth/client/tenant"
        dconfig "github.com/mendersoftware/deviceauth/config"
        "github.com/mendersoftware/deviceauth/model"
        "github.com/mendersoftware/deviceauth/store"
        "github.com/mendersoftware/deviceauth/store/mongo"
        "github.com/mendersoftware/deviceauth/utils"
)

var NowUnixMilis = utils.UnixMilis

func makeDataStoreConfig() mongo.DataStoreMongoConfig {
        return mongo.DataStoreMongoConfig{
                ConnectionString: config.Config.GetString(dconfig.SettingDb),

                SSL:           config.Config.GetBool(dconfig.SettingDbSSL),
                SSLSkipVerify: config.Config.GetBool(dconfig.SettingDbSSLSkipVerify),

                Username: config.Config.GetString(dconfig.SettingDbUsername),
                Password: config.Config.GetString(dconfig.SettingDbPassword),
        }

}

func Migrate(c config.Reader, tenant string, listTenantsFlag bool) error {
        db, err := mongo.NewDataStoreMongo(makeDataStoreConfig())

        if err != nil {
                return errors.Wrap(err, "failed to connect to db")
        }

        // list tenants only
        if listTenantsFlag {
                return listTenants(db)
        }

        db = db.WithAutomigrate().(*mongo.DataStoreMongo)

        if config.Config.Get(dconfig.SettingTenantAdmAddr) != "" {
                db = db.WithMultitenant()
        }

        ctx := context.Background()
        if tenant == "" {
                err = db.Migrate(ctx, mongo.DbVersion)
        } else {
                err = db.MigrateTenant(ctx, mongo.DbName, mongo.DbVersion)
                if err != nil {
                        return errors.Wrap(err, "failed to migrate main db")
                }

                tenantCtx := identity.WithContext(ctx, &identity.Identity{
                        Tenant: tenant,
                })
                dbname := mstore.DbFromContext(tenantCtx, mongo.DbName)
                err = db.MigrateTenant(tenantCtx, dbname, mongo.DbVersion)
        }
        if err != nil {
                return errors.Wrap(err, "failed to run migrations")
        }

        return nil
}

func listTenants(db *mongo.DataStoreMongo) error {
        tdbs, err := db.ListTenantsIds(context.Background())
        if err != nil {
                return errors.Wrap(err, "failed to retrieve tenant ids")
        }

        for _, tenant := range tdbs {
                fmt.Println(tenant)
        }

        return nil
}

func Maintenance(decommissioningCleanupFlag bool, tenant string, dryRunFlag bool) error {
        db, err := mongo.NewDataStoreMongo(makeDataStoreConfig())
        if err != nil {
                return errors.Wrap(err, "failed to connect to db")
        }

        return maintenanceWithDataStore(decommissioningCleanupFlag, tenant, dryRunFlag, db)
}

func maintenanceWithDataStore(
        decommissioningCleanupFlag bool,
        tenant string,
        dryRunFlag bool,
        db *mongo.DataStoreMongo,
) error {
        // cleanup devauth database from leftovers after failed decommissioning
        if decommissioningCleanupFlag {
                return decommissioningCleanup(db, tenant, dryRunFlag)
        }

        return nil
}

func decommissioningCleanup(db *mongo.DataStoreMongo, tenant string, dryRunFlag bool) error {
        if dryRunFlag {
                return decommissioningCleanupDryRun(db, tenant)
        } else {
                return decommissioningCleanupExecute(db, tenant)
        }
}

func decommissioningCleanupDryRun(db *mongo.DataStoreMongo, tenantId string) error {
        //devices
        devices, err := db.GetDevicesBeingDecommissioned(tenantId)
        if err != nil {
                return err
        }
        if len(devices) > 0 {
                fmt.Println("devices with decommissioning flag set:")
                for _, dev := range devices {
                        fmt.Println(dev.Id)
                }
        }

        //auth sets
        authSetIds, err := db.GetBrokenAuthSets(tenantId)
        if err != nil {
                return err
        }
        if len(authSetIds) > 0 {
                fmt.Println("authentication sets to be removed:")
                for _, authSetId := range authSetIds {
                        fmt.Println(authSetId)
                }
        }

        return nil
}

func decommissioningCleanupExecute(db *mongo.DataStoreMongo, tenantId string) error {
        if err := decommissioningCleanupDryRun(db, tenantId); err != nil {
                return err
        }

        if err := db.DeleteDevicesBeingDecommissioned(tenantId); err != nil {
                return err
        }

        if err := db.DeleteBrokenAuthSets(tenantId); err != nil {
                return err
        }

        return nil
}

func PropagateStatusesInventory(
        db store.DataStore,
        c cinv.Client,
        tenant string,
        migrationVersion string,
        dryRun bool,
) error {
        var err error

        l := log.NewEmpty()
        tenants := []string{tenant}
        if tenant == "" {
                tenants, err = db.ListTenantsIds(context.Background())
                if err != nil {
                        return errors.Wrap(err, "cant list tenants")
                }
        }

        var errReturned error
        for _, t := range tenants {
                err = tryPropagateStatusesInventoryForTenant(db, c, t, migrationVersion, dryRun)
                if err != nil {
                        errReturned = err
                        l.Errorf("giving up on tenant %s due to fatal error: %s", t, err.Error())
                        continue
                }
        }

        l.Info("all tenants processed, exiting.")
        return errReturned
}

func PropagateIdDataInventory(db store.DataStore, c cinv.Client, tenant string, dryRun bool) error {
        var err error

        l := log.NewEmpty()
        tenants := []string{tenant}
        if tenant == "" {
                tenants, err = db.ListTenantsIds(context.Background())
                if err != nil {
                        return errors.Wrap(err, "cant list tenants")
                }
        }

        var errReturned error
        for _, d := range tenants {
                err := tryPropagateIdDataInventoryForTenant(db, c, d, dryRun)
                if err != nil {
                        errReturned = err
                        l.Errorf("giving up on tenant %s due to fatal error: %s", d, err.Error())
                        continue
                }
        }

        l.Info("all tenants processed, exiting.")
        return errReturned
}

func PropagateReporting(
        db store.DataStore,
        wflows orchestrator.ClientRunner,
        tenant string,
        requestPeriod time.Duration,
        dryRun bool) error {
        l := log.NewEmpty()

        mapFunc := func(ctx context.Context) error {
                id := identity.FromContext(ctx)
                if id == nil || id.Tenant == "" {
                        // Not a tenant db - skip!
                        return nil
                }
                tenantId := id.Tenant
                return tryPropagateReportingForTenant(db, wflows, tenantId, requestPeriod, dryRun)
        }
        if tenant != "" {
                ctx := identity.WithContext(context.Background(),
                        &identity.Identity{
                                Tenant: tenant,
                        },
                )
                err := mapFunc(ctx)
                if err != nil {
                        return errors.Wrap(err, "failed to propagate for given tenant")
                }
                l.Infof("tenant processed, exiting.")
        } else {
                err := db.ForEachTenant(context.Background(), mapFunc)
                if err != nil {
                        return errors.Wrap(err, "failed to propagate for all tenant")
                }
                l.Info("all tenants processed, exiting.")
        }
        return nil
}

const (
        devicesBatchSize = 512
)

func updateDevicesStatus(
        ctx context.Context,
        db store.DataStore,
        c cinv.Client,
        tenant string,
        status string,
        dryRun bool,
) error {
        var skip uint

        skip = 0
        for {
                devices, err := db.GetDevices(ctx,
                        skip,
                        devicesBatchSize,
                        model.DeviceFilter{Status: []string{status}},
                )
                if err != nil {
                        return errors.Wrap(err, "failed to get devices")
                }

                if len(devices) < 1 {
                        break
                }

                deviceUpdates := make([]model.DeviceInventoryUpdate, len(devices))

                for i, d := range devices {
                        deviceUpdates[i].Id = d.Id
                        deviceUpdates[i].Revision = d.Revision
                }

                if !dryRun {
                        err = c.SetDeviceStatus(ctx, tenant, deviceUpdates, status)
                        if err != nil {
                                return err
                        }
                }

                if len(devices) < devicesBatchSize {
                        break
                } else {
                        skip += devicesBatchSize
                }
        }
        return nil
}

func updateDevicesIdData(
        ctx context.Context,
        db store.DataStore,
        c cinv.Client,
        tenant string,
        dryRun bool,
) error {
        var skip uint

        skip = 0
        for {
                devices, err := db.GetDevices(ctx, skip, devicesBatchSize, model.DeviceFilter{})
                if err != nil {
                        return errors.Wrap(err, "failed to get devices")
                }

                if len(devices) < 1 {
                        break
                }

                if !dryRun {
                        for _, d := range devices {
                                err := c.SetDeviceIdentity(ctx, tenant, d.Id, d.IdDataStruct)
                                if err != nil {
                                        return err
                                }
                        }
                }

                skip += devicesBatchSize
                if len(devices) < devicesBatchSize {
                        break
                }
        }
        return nil
}

func tryPropagateStatusesInventoryForTenant(
        db store.DataStore,
        c cinv.Client,
        tenant string,
        migrationVersion string,
        dryRun bool,
) error {
        l := log.NewEmpty()

        l.Infof("propagating device statuses to inventory from tenant: %s", tenant)

        ctx := context.Background()
        if tenant != "" {
                ctx = identity.WithContext(ctx, &identity.Identity{
                        Tenant: tenant,
                })
        }

        var err error
        var errReturned error
        for _, status := range model.DevStatuses {
                err = updateDevicesStatus(ctx, db, c, tenant, status, dryRun)
                if err != nil {
                        l.Infof(
                                "Done with tenant %s status=%s, but there were errors: %s.",
                                tenant,
                                status,
                                err.Error(),
                        )
                        errReturned = err
                } else {
                        l.Infof("Done with tenant %s status=%s", tenant, status)
                }
        }
        if migrationVersion != "" && !dryRun {
                if errReturned != nil {
                        l.Warnf(
                                "Will not store %s migration version for tenant %s due to errors.",
                                migrationVersion,
                                tenant,
                        )
                } else {
                        version, err := migrate.NewVersion(migrationVersion)
                        if version == nil || err != nil {
                                l.Warnf(
                                        "Will not store %s migration version in %s.migration_info due to bad version"+
                                                " provided.",
                                        migrationVersion,
                                        tenant,
                                )
                                errReturned = err
                        } else {
                                _ = db.StoreMigrationVersion(ctx, version)
                        }
                }
        }

        return errReturned
}

func tryPropagateIdDataInventoryForTenant(
        db store.DataStore,
        c cinv.Client,
        tenant string,
        dryRun bool,
) error {
        l := log.NewEmpty()

        l.Infof("propagating device id_data to inventory from tenant: %s", tenant)

        ctx := context.Background()
        if tenant != "" {
                ctx = identity.WithContext(ctx, &identity.Identity{
                        Tenant: tenant,
                })
        }

        err := updateDevicesIdData(ctx, db, c, tenant, dryRun)
        if err != nil {
                l.Infof("Done with tenant %s, but there were errors: %s.", tenant, err.Error())
        } else {
                l.Infof("Done with tenant %s", tenant)
        }

        return err
}

func tryPropagateReportingForTenant(
        db store.DataStore,
        wflows orchestrator.ClientRunner,
        tenant string,
        requestPeriod time.Duration,
        dryRun bool,
) error {
        l := log.NewEmpty()

        l.Infof("propagating device data to reporting for tenant %s", tenant)

        ctx := context.Background()
        if tenant != "" {
                ctx = identity.WithContext(ctx, &identity.Identity{
                        Tenant: tenant,
                })
        } else {
                return errors.New("you must provide a tenant id")
        }

        err := reindexDevicesReporting(ctx, requestPeriod, db, wflows, dryRun)
        if err != nil {
                l.Infof("Done with tenant %s, but there were errors: %s.", tenant, err.Error())
        } else {
                l.Infof("Done with tenant %s", tenant)
        }

        return err
}

func reindexDevicesReporting(
        ctx context.Context,
        requestPeriod time.Duration,
        db store.DataStore,
        wflows orchestrator.ClientRunner,
        dryRun bool,
) error {
        var skip uint

        skip = 0
        done := ctx.Done()
        rateLimit := time.NewTicker(requestPeriod)
        defer rateLimit.Stop()
        for {
                devices, err := db.GetDevices(ctx, skip, devicesBatchSize, model.DeviceFilter{})
                if err != nil {
                        return errors.Wrap(err, "failed to get devices")
                }

                if len(devices) < 1 {
                        break
                }

                if !dryRun {
                        deviceIDs := make([]string, len(devices))
                        for i, d := range devices {
                                deviceIDs[i] = d.Id
                        }
                        err := wflows.SubmitReindexReportingBatch(ctx, deviceIDs)
                        if err != nil {
                                return err
                        }
                }

                skip += devicesBatchSize
                if len(devices) < devicesBatchSize {
                        break
                }
                select {
                case <-rateLimit.C:

                case <-done:
                        return ctx.Err()
                }
        }
        return nil
}

const (
        WorkflowsDeviceLimitText    = "@/etc/workflows-enterprise/data/device_limit_email.txt"
        WorkflowsDeviceLimitHTML    = "@/etc/workflows-enterprise/data/device_limit_email.html"
        WorkflowsDeviceLimitSubject = "Device limit almost reached"
)

func warnTenantUsers(
        ctx context.Context,
        tenantID string,
        tadm tenant.ClientRunner,
        wflows orchestrator.ClientRunner,
        remainingDevices uint,
) error {
        users, err := tadm.GetTenantUsers(ctx, tenantID)
        if err != nil {
                // Log the event and continue with the other tenants
                return err
        }
        for i := range users {
                warnWFlow := orchestrator.DeviceLimitWarning{
                        RequestID:      "deviceAuthAdmin",
                        RecipientEmail: users[i].Email,

                        Subject:          WorkflowsDeviceLimitSubject,
                        Body:             WorkflowsDeviceLimitText,
                        BodyHTML:         WorkflowsDeviceLimitHTML,
                        RemainingDevices: &remainingDevices,
                }
                err = wflows.SubmitDeviceLimitWarning(ctx, warnWFlow)
                if err != nil {
                        return err
                }
        }
        return nil
}

// CheckDeviceLimits goes through all tenant databases and checks if the number
// of accepted devices is above a given threshold (in %) and sends an email
// to all registered users registered under the given tenant.
func CheckDeviceLimits(
        threshold float64,
        ds store.DataStore,
        tadm tenant.ClientRunner,
        wflows orchestrator.ClientRunner,
) error {
        // Sanitize threshold
        if threshold > 100.0 {
                threshold = 100.0
        } else if threshold < 0.0 {
                threshold = 0.0
        }
        threshProportion := threshold / 100.0

        // mapFunc is applied to all existing databases in datastore.
        mapFunc := func(ctx context.Context) error {
                id := identity.FromContext(ctx)
                if id == nil || id.Tenant == "" {
                        // Not a tenant db - skip!
                        return nil
                }
                tenantID := id.Tenant
                l := log.FromContext(ctx)

                lim, err := ds.GetLimit(ctx, model.LimitMaxDeviceCount)
                if err != nil {
                        return err
                }
                n, err := ds.GetDevCountByStatus(ctx, model.DevStatusAccepted)
                if err != nil {
                        return err
                }
                if float64(n) >= (float64(lim.Value) * threshProportion) {
                        // User is above limit

                        remainingUsers := uint(n) - uint(lim.Value)
                        err := warnTenantUsers(ctx, tenantID, tadm, wflows, remainingUsers)
                        if err != nil {
                                l.Warnf(`Failed to warn tenant "%s" `+
                                        `users nearing device limit: %s`,
                                        tenantID, err.Error(),
                                )
                        }
                }
                return nil
        }
        // Start looping through the databases.
        return ds.ForEachTenant(
                context.Background(),
                mapFunc,
        )
}

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	package cmd
15
16	import (
17	"context"
18	"fmt"
19	"time"
20
21	"github.com/mendersoftware/go-lib-micro/config"
22	"github.com/mendersoftware/go-lib-micro/identity"
23	"github.com/mendersoftware/go-lib-micro/log"
24	"github.com/mendersoftware/go-lib-micro/mongo/migrate"
25	mstore "github.com/mendersoftware/go-lib-micro/store"
26	"github.com/pkg/errors"
27
28	cinv "github.com/mendersoftware/deviceauth/client/inventory"
29	"github.com/mendersoftware/deviceauth/client/orchestrator"
30	"github.com/mendersoftware/deviceauth/client/tenant"
31	dconfig "github.com/mendersoftware/deviceauth/config"
32	"github.com/mendersoftware/deviceauth/model"
33	"github.com/mendersoftware/deviceauth/store"
34	"github.com/mendersoftware/deviceauth/store/mongo"
35	"github.com/mendersoftware/deviceauth/utils"
36	)
37
38	var NowUnixMilis = utils.UnixMilis
39
40	func makeDataStoreConfig() mongo.DataStoreMongoConfig {	1✔
41	return mongo.DataStoreMongoConfig{	1✔
42	ConnectionString: config.Config.GetString(dconfig.SettingDb),	1✔
43		1✔
44	SSL: config.Config.GetBool(dconfig.SettingDbSSL),	1✔
45	SSLSkipVerify: config.Config.GetBool(dconfig.SettingDbSSLSkipVerify),	1✔
46		1✔
47	Username: config.Config.GetString(dconfig.SettingDbUsername),	1✔
48	Password: config.Config.GetString(dconfig.SettingDbPassword),	1✔
49	}	1✔
50		1✔
51	}	1✔
52
53	func Migrate(c config.Reader, tenant string, listTenantsFlag bool) error {	1✔
54	db, err := mongo.NewDataStoreMongo(makeDataStoreConfig())	1✔
55		1✔
56	if err != nil {	1✔
57	return errors.Wrap(err, "failed to connect to db")	×
58	}	×
59
60	// list tenants only
61	if listTenantsFlag {	1✔
62	return listTenants(db)	×
63	}	×
64
65	db = db.WithAutomigrate().(*mongo.DataStoreMongo)	1✔
66		1✔
67	if config.Config.Get(dconfig.SettingTenantAdmAddr) != "" {	1✔
68	db = db.WithMultitenant()	×
69	}	×
70
71	ctx := context.Background()	1✔
72	if tenant == "" {	2✔
73	err = db.Migrate(ctx, mongo.DbVersion)	1✔
74	} else {	1✔
75	err = db.MigrateTenant(ctx, mongo.DbName, mongo.DbVersion)	×
76	if err != nil {	×
77	return errors.Wrap(err, "failed to migrate main db")	×
78	}	×
79
80	tenantCtx := identity.WithContext(ctx, &identity.Identity{	×
81	Tenant: tenant,	×
82	})	×
83	dbname := mstore.DbFromContext(tenantCtx, mongo.DbName)	×
84	err = db.MigrateTenant(tenantCtx, dbname, mongo.DbVersion)	×
85	}
86	if err != nil {	1✔
87	return errors.Wrap(err, "failed to run migrations")	×
88	}	×
89
90	return nil	1✔
91	}
92
93	func listTenants(db *mongo.DataStoreMongo) error {	×
94	tdbs, err := db.ListTenantsIds(context.Background())	×
95	if err != nil {	×
96	return errors.Wrap(err, "failed to retrieve tenant ids")	×
97	}	×
98
99	for _, tenant := range tdbs {	×
100	fmt.Println(tenant)	×
101	}	×
102
103	return nil	×
104	}
105
106	func Maintenance(decommissioningCleanupFlag bool, tenant string, dryRunFlag bool) error {	×
107	db, err := mongo.NewDataStoreMongo(makeDataStoreConfig())	×
108	if err != nil {	×
109	return errors.Wrap(err, "failed to connect to db")	×
110	}	×
111
112	return maintenanceWithDataStore(decommissioningCleanupFlag, tenant, dryRunFlag, db)	×
113	}
114
115	func maintenanceWithDataStore(
116	decommissioningCleanupFlag bool,
117	tenant string,
118	dryRunFlag bool,
119	db *mongo.DataStoreMongo,
120	) error {	8✔
121	// cleanup devauth database from leftovers after failed decommissioning	8✔
122	if decommissioningCleanupFlag {	14✔
123	return decommissioningCleanup(db, tenant, dryRunFlag)	6✔
124	}	6✔
125
126	return nil	2✔
127	}
128
129	func decommissioningCleanup(db *mongo.DataStoreMongo, tenant string, dryRunFlag bool) error {	6✔
130	if dryRunFlag {	9✔
131	return decommissioningCleanupDryRun(db, tenant)	3✔
132	} else {	6✔
133	return decommissioningCleanupExecute(db, tenant)	3✔
134	}	3✔
135	}
136
137	func decommissioningCleanupDryRun(db *mongo.DataStoreMongo, tenantId string) error {	6✔
138	//devices	6✔
139	devices, err := db.GetDevicesBeingDecommissioned(tenantId)	6✔
140	if err != nil {	6✔
141	return err	×
142	}	×
143	if len(devices) > 0 {	8✔
144	fmt.Println("devices with decommissioning flag set:")	2✔
145	for _, dev := range devices {	4✔
146	fmt.Println(dev.Id)	2✔
147	}	2✔
148	}
149
150	//auth sets
151	authSetIds, err := db.GetBrokenAuthSets(tenantId)	6✔
152	if err != nil {	6✔
153	return err	×
154	}	×
155	if len(authSetIds) > 0 {	8✔
156	fmt.Println("authentication sets to be removed:")	2✔
157	for _, authSetId := range authSetIds {	4✔
158	fmt.Println(authSetId)	2✔
159	}	2✔
160	}
161
162	return nil	6✔
163	}
164
165	func decommissioningCleanupExecute(db *mongo.DataStoreMongo, tenantId string) error {	3✔
166	if err := decommissioningCleanupDryRun(db, tenantId); err != nil {	3✔
167	return err	×
168	}	×
169
170	if err := db.DeleteDevicesBeingDecommissioned(tenantId); err != nil {	3✔
171	return err	×
172	}	×
173
174	if err := db.DeleteBrokenAuthSets(tenantId); err != nil {	3✔
175	return err	×
176	}	×
177
178	return nil	3✔
179	}
180
181	func PropagateStatusesInventory(
182	db store.DataStore,
183	c cinv.Client,
184	tenant string,
185	migrationVersion string,
186	dryRun bool,
187	) error {	6✔
188	var err error	6✔
189		6✔
190	l := log.NewEmpty()	6✔
191	tenants := []string{tenant}	6✔
192	if tenant == "" {	10✔
193	tenants, err = db.ListTenantsIds(context.Background())	4✔
194	if err != nil {	5✔
195	return errors.Wrap(err, "cant list tenants")	1✔
196	}	1✔
197	}
198
199	var errReturned error	5✔
200	for _, t := range tenants {	16✔
201	err = tryPropagateStatusesInventoryForTenant(db, c, t, migrationVersion, dryRun)	11✔
202	if err != nil {	18✔
203	errReturned = err	7✔
204	l.Errorf("giving up on tenant %s due to fatal error: %s", t, err.Error())	7✔
205	continue	7✔
206	}
207	}
208
209	l.Info("all tenants processed, exiting.")	5✔
210	return errReturned	5✔
211	}
212
213	func PropagateIdDataInventory(db store.DataStore, c cinv.Client, tenant string, dryRun bool) error {	×
214	var err error	×
215		×
216	l := log.NewEmpty()	×
217	tenants := []string{tenant}	×
218	if tenant == "" {	×
219	tenants, err = db.ListTenantsIds(context.Background())	×
220	if err != nil {	×
221	return errors.Wrap(err, "cant list tenants")	×
222	}	×
223	}
224
225	var errReturned error	×
226	for _, d := range tenants {	×
227	err := tryPropagateIdDataInventoryForTenant(db, c, d, dryRun)	×
228	if err != nil {	×
229	errReturned = err	×
230	l.Errorf("giving up on tenant %s due to fatal error: %s", d, err.Error())	×
231	continue	×
232	}
233	}
234
235	l.Info("all tenants processed, exiting.")	×
236	return errReturned	×
237	}
238
239	func PropagateReporting(
240	db store.DataStore,
241	wflows orchestrator.ClientRunner,
242	tenant string,
243	requestPeriod time.Duration,
244	dryRun bool) error {	4✔
245	l := log.NewEmpty()	4✔
246		4✔
247	mapFunc := func(ctx context.Context) error {	12✔
248	id := identity.FromContext(ctx)	8✔
249	if id == nil \|\| id.Tenant == "" {	8✔
250	// Not a tenant db - skip!	×
251	return nil	×
252	}	×
253	tenantId := id.Tenant	8✔
254	return tryPropagateReportingForTenant(db, wflows, tenantId, requestPeriod, dryRun)	8✔
255	}
256	if tenant != "" {	4✔
257	ctx := identity.WithContext(context.Background(),	×
258	&identity.Identity{	×
259	Tenant: tenant,	×
260	},	×
261	)	×
262	err := mapFunc(ctx)	×
263	if err != nil {	×
264	return errors.Wrap(err, "failed to propagate for given tenant")	×
265	}	×
266	l.Infof("tenant processed, exiting.")	×
267	} else {	4✔
268	err := db.ForEachTenant(context.Background(), mapFunc)	4✔
269	if err != nil {	4✔
270	return errors.Wrap(err, "failed to propagate for all tenant")	×
271	}	×
272	l.Info("all tenants processed, exiting.")	4✔
273	}
274	return nil	4✔
275	}
276
277	const (
278	devicesBatchSize = 512
279	)
280
281	func updateDevicesStatus(
282	ctx context.Context,
283	db store.DataStore,
284	c cinv.Client,
285	tenant string,
286	status string,
287	dryRun bool,
288	) error {	55✔
289	var skip uint	55✔
290		55✔
291	skip = 0	55✔
292	for {	110✔
293	devices, err := db.GetDevices(ctx,	55✔
294	skip,	55✔
295	devicesBatchSize,	55✔
296	model.DeviceFilter{Status: []string{status}},	55✔
297	)	55✔
298	if err != nil {	70✔
299	return errors.Wrap(err, "failed to get devices")	15✔
300	}	15✔
301
302	if len(devices) < 1 {	40✔
303	break	×
304	}
305
306	deviceUpdates := make([]model.DeviceInventoryUpdate, len(devices))	40✔
307		40✔
308	for i, d := range devices {	120✔
309	deviceUpdates[i].Id = d.Id	80✔
310	deviceUpdates[i].Revision = d.Revision	80✔
311	}	80✔
312
313	if !dryRun {	80✔
314	err = c.SetDeviceStatus(ctx, tenant, deviceUpdates, status)	40✔
315	if err != nil {	55✔
316	return err	15✔
317	}	15✔
318	}
319
320	if len(devices) < devicesBatchSize {	50✔
321	break	25✔
322	} else {	×
323	skip += devicesBatchSize	×
324	}	×
325	}
326	return nil	25✔
327	}
328
329	func updateDevicesIdData(
330	ctx context.Context,
331	db store.DataStore,
332	c cinv.Client,
333	tenant string,
334	dryRun bool,
335	) error {	×
336	var skip uint	×
337		×
338	skip = 0	×
339	for {	×
340	devices, err := db.GetDevices(ctx, skip, devicesBatchSize, model.DeviceFilter{})	×
341	if err != nil {	×
342	return errors.Wrap(err, "failed to get devices")	×
343	}	×
344
345	if len(devices) < 1 {	×
346	break	×
347	}
348
349	if !dryRun {	×
350	for _, d := range devices {	×
351	err := c.SetDeviceIdentity(ctx, tenant, d.Id, d.IdDataStruct)	×
352	if err != nil {	×
353	return err	×
354	}	×
355	}
356	}
357
358	skip += devicesBatchSize	×
359	if len(devices) < devicesBatchSize {	×
360	break	×
361	}
362	}
363	return nil	×
364	}
365
366	func tryPropagateStatusesInventoryForTenant(
367	db store.DataStore,
368	c cinv.Client,
369	tenant string,
370	migrationVersion string,
371	dryRun bool,
372	) error {	11✔
373	l := log.NewEmpty()	11✔
374		11✔
375	l.Infof("propagating device statuses to inventory from tenant: %s", tenant)	11✔
376		11✔
377	ctx := context.Background()	11✔
378	if tenant != "" {	22✔
379	ctx = identity.WithContext(ctx, &identity.Identity{	11✔
380	Tenant: tenant,	11✔
381	})	11✔
382	}	11✔
383
384	var err error	11✔
385	var errReturned error	11✔
386	for _, status := range model.DevStatuses {	66✔
387	err = updateDevicesStatus(ctx, db, c, tenant, status, dryRun)	55✔
388	if err != nil {	85✔
389	l.Infof(	30✔
390	"Done with tenant %s status=%s, but there were errors: %s.",	30✔
391	tenant,	30✔
392	status,	30✔
393	err.Error(),	30✔
394	)	30✔
395	errReturned = err	30✔
396	} else {	55✔
397	l.Infof("Done with tenant %s status=%s", tenant, status)	25✔
398	}	25✔
399	}
400	if migrationVersion != "" && !dryRun {	13✔
401	if errReturned != nil {	2✔
402	l.Warnf(	×
403	"Will not store %s migration version for tenant %s due to errors.",	×
404	migrationVersion,	×
405	tenant,	×
406	)	×
407	} else {	2✔
408	version, err := migrate.NewVersion(migrationVersion)	2✔
409	if version == nil \|\| err != nil {	3✔
410	l.Warnf(	1✔
411	"Will not store %s migration version in %s.migration_info due to bad version"+	1✔
412	" provided.",	1✔
413	migrationVersion,	1✔
414	tenant,	1✔
415	)	1✔
416	errReturned = err	1✔
417	} else {	2✔
418	_ = db.StoreMigrationVersion(ctx, version)	1✔
419	}	1✔
420	}
421	}
422
423	return errReturned	11✔
424	}
425
426	func tryPropagateIdDataInventoryForTenant(
427	db store.DataStore,
428	c cinv.Client,
429	tenant string,
430	dryRun bool,
431	) error {	×
432	l := log.NewEmpty()	×
433		×
434	l.Infof("propagating device id_data to inventory from tenant: %s", tenant)	×
435		×
436	ctx := context.Background()	×
437	if tenant != "" {	×
438	ctx = identity.WithContext(ctx, &identity.Identity{	×
439	Tenant: tenant,	×
440	})	×
441	}	×
442
443	err := updateDevicesIdData(ctx, db, c, tenant, dryRun)	×
444	if err != nil {	×
445	l.Infof("Done with tenant %s, but there were errors: %s.", tenant, err.Error())	×
446	} else {	×
447	l.Infof("Done with tenant %s", tenant)	×
448	}	×
449
450	return err	×
451	}
452
453	func tryPropagateReportingForTenant(
454	db store.DataStore,
455	wflows orchestrator.ClientRunner,
456	tenant string,
457	requestPeriod time.Duration,
458	dryRun bool,
459	) error {	8✔
460	l := log.NewEmpty()	8✔
461		8✔
462	l.Infof("propagating device data to reporting for tenant %s", tenant)	8✔
463		8✔
464	ctx := context.Background()	8✔
465	if tenant != "" {	16✔
466	ctx = identity.WithContext(ctx, &identity.Identity{	8✔
467	Tenant: tenant,	8✔
468	})	8✔
469	} else {	8✔
470	return errors.New("you must provide a tenant id")	×
471	}	×
472
473	err := reindexDevicesReporting(ctx, requestPeriod, db, wflows, dryRun)	8✔
474	if err != nil {	10✔
475	l.Infof("Done with tenant %s, but there were errors: %s.", tenant, err.Error())	2✔
476	} else {	8✔
477	l.Infof("Done with tenant %s", tenant)	6✔
478	}	6✔
479
480	return err	8✔
481	}
482
483	func reindexDevicesReporting(
484	ctx context.Context,
485	requestPeriod time.Duration,
486	db store.DataStore,
487	wflows orchestrator.ClientRunner,
488	dryRun bool,
489	) error {	8✔
490	var skip uint	8✔
491		8✔
492	skip = 0	8✔
493	done := ctx.Done()	8✔
494	rateLimit := time.NewTicker(requestPeriod)	8✔
495	defer rateLimit.Stop()	8✔
496	for {	16✔
497	devices, err := db.GetDevices(ctx, skip, devicesBatchSize, model.DeviceFilter{})	8✔
498	if err != nil {	9✔
499	return errors.Wrap(err, "failed to get devices")	1✔
500	}	1✔
501
502	if len(devices) < 1 {	7✔
503	break	×
504	}
505
506	if !dryRun {	11✔
507	deviceIDs := make([]string, len(devices))	4✔
508	for i, d := range devices {	11✔
509	deviceIDs[i] = d.Id	7✔
510	}	7✔
511	err := wflows.SubmitReindexReportingBatch(ctx, deviceIDs)	4✔
512	if err != nil {	5✔
513	return err	1✔
514	}	1✔
515	}
516
517	skip += devicesBatchSize	6✔
518	if len(devices) < devicesBatchSize {	12✔
519	break	6✔
520	}
521	select {	×
522	case <-rateLimit.C:	×
523
524	case <-done:	×
525	return ctx.Err()	×
526	}
527	}
528	return nil	6✔
529	}
530
531	const (
532	WorkflowsDeviceLimitText = "@/etc/workflows-enterprise/data/device_limit_email.txt"
533	WorkflowsDeviceLimitHTML = "@/etc/workflows-enterprise/data/device_limit_email.html"
534	WorkflowsDeviceLimitSubject = "Device limit almost reached"
535	)
536
537	func warnTenantUsers(
538	ctx context.Context,
539	tenantID string,
540	tadm tenant.ClientRunner,
541	wflows orchestrator.ClientRunner,
542	remainingDevices uint,
543	) error {	5✔
544	users, err := tadm.GetTenantUsers(ctx, tenantID)	5✔
545	if err != nil {	6✔
546	// Log the event and continue with the other tenants	1✔
547	return err	1✔
548	}	1✔
549	for i := range users {	11✔
550	warnWFlow := orchestrator.DeviceLimitWarning{	7✔
551	RequestID: "deviceAuthAdmin",	7✔
552	RecipientEmail: users[i].Email,	7✔
553		7✔
554	Subject: WorkflowsDeviceLimitSubject,	7✔
555	Body: WorkflowsDeviceLimitText,	7✔
556	BodyHTML: WorkflowsDeviceLimitHTML,	7✔
557	RemainingDevices: &remainingDevices,	7✔
558	}	7✔
559	err = wflows.SubmitDeviceLimitWarning(ctx, warnWFlow)	7✔
560	if err != nil {	8✔
561	return err	1✔
562	}	1✔
563	}
564	return nil	3✔
565	}
566
567	// CheckDeviceLimits goes through all tenant databases and checks if the number
568	// of accepted devices is above a given threshold (in %) and sends an email
569	// to all registered users registered under the given tenant.
570	func CheckDeviceLimits(
571	threshold float64,
572	ds store.DataStore,
573	tadm tenant.ClientRunner,
574	wflows orchestrator.ClientRunner,
575	) error {	5✔
576	// Sanitize threshold	5✔
577	if threshold > 100.0 {	6✔
578	threshold = 100.0	1✔
579	} else if threshold < 0.0 {	6✔
580	threshold = 0.0	1✔
581	}	1✔
582	threshProportion := threshold / 100.0	5✔
583		5✔
584	// mapFunc is applied to all existing databases in datastore.	5✔
585	mapFunc := func(ctx context.Context) error {	17✔
586	id := identity.FromContext(ctx)	12✔
587	if id == nil \|\| id.Tenant == "" {	13✔
588	// Not a tenant db - skip!	1✔
589	return nil	1✔
590	}	1✔
591	tenantID := id.Tenant	11✔
592	l := log.FromContext(ctx)	11✔
593		11✔
594	lim, err := ds.GetLimit(ctx, model.LimitMaxDeviceCount)	11✔
595	if err != nil {	12✔
596	return err	1✔
597	}	1✔
598	n, err := ds.GetDevCountByStatus(ctx, model.DevStatusAccepted)	10✔
599	if err != nil {	11✔
600	return err	1✔
601	}	1✔
602	if float64(n) >= (float64(lim.Value) * threshProportion) {	14✔
603	// User is above limit	5✔
604		5✔
605	remainingUsers := uint(n) - uint(lim.Value)	5✔
606	err := warnTenantUsers(ctx, tenantID, tadm, wflows, remainingUsers)	5✔
607	if err != nil {	7✔
608	l.Warnf(`Failed to warn tenant "%s" `+	2✔
609	`users nearing device limit: %s`,	2✔
610	tenantID, err.Error(),	2✔
611	)	2✔
612	}	2✔
613	}
614	return nil	9✔
615	}
616	// Start looping through the databases.
617	return ds.ForEachTenant(	5✔
618	context.Background(),	5✔
619	mapFunc,	5✔
620	)	5✔
621	}

mendersoftware / deviceauth / 1284672998

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