760051305

Build Type

gitlab-ci

Committed by Fabio Tranchitella

Commit Message

feat: CLI command to reindex to the reporting service all the devices

Pull Request Pull Request #625: feat: CLI command to reindex to the reporting service all the devices

Run Details

59 of 144 new or added lines in 3 files covered. (40.97%)

7 existing lines in 2 files now uncovered.

4466 of 5370 relevant lines covered (83.17%)

47.73 hits per line

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

63.47

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

        "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 {
                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.GetTenantDbs()
        if err != nil {
                return errors.Wrap(err, "failed to retrieve tenant DBs")
        }

        for _, tenant := range tdbs {
                fmt.Println(mstore.TenantFromDbName(tenant, mongo.DbName))
        }

        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 tenant == "" {
                tdbs, err := db.GetTenantDbs()
                if err != nil {
                        return errors.Wrap(err, "failed to retrieve tenant DBs")
                }
                _ = decommissioningCleanupWithDbs(db, append(tdbs, mongo.DbName), dryRunFlag)
        } else {
                _ = decommissioningCleanupWithDbs(
                        db,
                        []string{mstore.DbNameForTenant(tenant, mongo.DbName)},
                        dryRunFlag,
                )
        }

        return nil
}

func decommissioningCleanupWithDbs(
        db *mongo.DataStoreMongo,
        tenantDbs []string,
        dryRunFlag bool,
) error {
        for _, dbName := range tenantDbs {
                println("database: ", dbName)
                if err := decommissioningCleanupWithDb(db, dbName, dryRunFlag); err != nil {
                        return err
                }
        }
        return nil
}

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

func decommissioningCleanupDryRun(db *mongo.DataStoreMongo, dbName string) error {
        //devices
        devices, err := db.GetDevicesBeingDecommissioned(dbName)
        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(dbName)
        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, dbName string) error {
        if err := decommissioningCleanupDryRun(db, dbName); err != nil {
                return err
        }

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

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

        return nil
}

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

        dbs, err := selectDbs(db, tenant)
        if err != nil {
                return errors.Wrap(err, "aborting")
        }

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

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

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

        dbs, err := selectDbs(db, tenant)
        if err != nil {
                return errors.Wrap(err, "aborting")
        }

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

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

func PropagateReporting(db store.DataStore, wflows *orchestrator.Client, tenant string,
        dryRun bool) error {
        l := log.NewEmpty()

        dbs, err := selectDbs(db, tenant)
        if err != nil {
                return errors.Wrap(err, "aborting")
        }

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

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

func selectDbs(db store.DataStore, tenant string) ([]string, error) {
        l := log.NewEmpty()

        var dbs []string

        if tenant != "" {
                l.Infof("propagating inventory for user-specified tenant %s", tenant)
                n := mstore.DbNameForTenant(tenant, mongo.DbName)
                dbs = []string{n}
        } else {
                l.Infof("propagating inventory for all tenants")

                // infer if we're in ST or MT
                tdbs, err := db.GetTenantDbs()
                if err != nil {
                        return nil, errors.Wrap(err, "failed to retrieve tenant DBs")
                }

                if len(tdbs) == 0 {
                        l.Infof("no tenant DBs found - will try the default database %s", mongo.DbName)
                        dbs = []string{mongo.DbName}
                } else {
                        dbs = tdbs
                }
        }

        return dbs, 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
                                }
                        }
                }

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

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

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

        tenant := mstore.TenantFromDbName(dbname, mongo.DbName)

        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 DB %s status=%s, but there were errors: %s.",
                                dbname,
                                status,
                                err.Error(),
                        )
                        errReturned = err
                } else {
                        l.Infof("Done with DB %s status=%s", dbname, status)
                }
        }
        if migrationVersion != "" && !dryRun {
                if errReturned != nil {
                        l.Warnf(
                                "Will not store %s migration version in %s.migration_info due to errors.",
                                migrationVersion,
                                dbname,
                        )
                } 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,
                                        dbname,
                                )
                                errReturned = err
                        } else {
                                _ = db.StoreMigrationVersion(ctx, version)
                        }
                }
        }

        return errReturned
}

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

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

        tenant := mstore.TenantFromDbName(dbname, mongo.DbName)

        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 DB %s, but there were errors: %s.", dbname, err.Error())
        } else {
                l.Infof("Done with DB %s", dbname)
        }

        return err
}

func tryPropagateReportingForDb(
        db store.DataStore,
        wflows *orchestrator.Client,
        dbname string,
        dryRun bool,
) error {
        l := log.NewEmpty()

        l.Infof("propagating device data to reporting from DB: %s", dbname)

        tenant := mstore.TenantFromDbName(dbname, mongo.DbName)

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

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

        return err
}

func reindexDevicesReporting(
        ctx context.Context,
        db store.DataStore,
        wflows *orchestrator.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 {
                        deviceIDs := make([]string, len(devices))
                        for i, d := range devices {
                                deviceIDs[i] = d.Id
                        }
                        err := wflows.SubmitReindexReportingBatch(ctx, deviceIDs)
                        if err != nil {
                                return err
                        }
                }

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

mendersoftware / deviceauth / 760051305

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