850112129

Build Type

gitlab-ci

Committed by Peter Grzybowski

Commit Message

chore: migrate dbName upfront in case of migrating a tenant.

Pull Request Pull Request #642: Single db move fixes

Run Details

332 of 405 new or added lines in 5 files covered. (81.98%)

89 existing lines in 3 files now uncovered.

4672 of 5591 relevant lines covered (83.56%)

79.43 hits per line

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

70.62

/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 {
                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 {
        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.ClientRunner, tenant string,
        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, 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
}

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

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

                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 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 tryPropagateReportingForTenant(
        db store.DataStore,
        wflows orchestrator.ClientRunner,
        tenant string,
        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, 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,
        db store.DataStore,
        wflows orchestrator.ClientRunner,
        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
                        }
                }

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

mendersoftware / deviceauth / 850112129

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