948596009

pending completion

Build # 948596009

Build Type

push

gitlab-ci

Committed by

web-flow

Commit Message

Merge pull request #655 from merlin-northern/men_6529_merge_single_db_to_master

Merge single db to master

Run Details

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

16 existing lines in 2 files now uncovered.

4809 of 5767 relevant lines covered (83.39%)

48.84 hits per line

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

70.15

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

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,
        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 {	2✔
62	return listTenants(db)	1✔
63	}	1✔
64
65	db = db.WithAutomigrate().(*mongo.DataStoreMongo)	1✔
66		1✔
67	if config.Config.Get(dconfig.SettingTenantAdmAddr) != "" {	2✔
68	db = db.WithMultitenant()	1✔
69	}	1✔
70
71	ctx := context.Background()	1✔
72	if tenant == "" {	2✔
73	err = db.Migrate(ctx, mongo.DbVersion)	1✔
74	} else {	2✔
75	err = db.MigrateTenant(ctx, mongo.DbName, mongo.DbVersion)	1✔
76	if err != nil {	1✔
NEW 77	return errors.Wrap(err, "failed to migrate main db")	×
NEW 78	}	×
79
80	tenantCtx := identity.WithContext(ctx, &identity.Identity{	1✔
81	Tenant: tenant,	1✔
82	})	1✔
83	dbname := mstore.DbFromContext(tenantCtx, mongo.DbName)	1✔
84	err = db.MigrateTenant(tenantCtx, dbname, mongo.DbVersion)	1✔
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 {	1✔
94	tdbs, err := db.ListTenantsIds(context.Background())	1✔
95	if err != nil {	1✔
NEW 96	return errors.Wrap(err, "failed to retrieve tenant ids")	×
97	}	×
98
99	for _, tenant := range tdbs {	2✔
100	fmt.Println(tenant)	1✔
101	}	1✔
102
103	return nil	1✔
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	l := log.NewEmpty()	×
215		×
216	dbs, err := selectDbs(db, tenant)	×
217	if err != nil {	×
218	return errors.Wrap(err, "aborting")	×
219	}	×
220
221	var errReturned error	×
222	for _, d := range dbs {	×
223	err := tryPropagateIdDataInventoryForDb(db, c, d, dryRun)	×
224	if err != nil {	×
225	errReturned = err	×
226	l.Errorf("giving up on DB %s due to fatal error: %s", d, err.Error())	×
227	continue	×
228	}
229	}
230
231	l.Info("all DBs processed, exiting.")	×
232	return errReturned	×
233	}
234
235	func PropagateReporting(
236	db store.DataStore,
237	wflows orchestrator.ClientRunner,
238	tenant string,
239	requestPeriod time.Duration,
240	dryRun bool) error {	4✔
241	l := log.NewEmpty()	4✔
242		4✔
243	mapFunc := func(ctx context.Context) error {	12✔
244	id := identity.FromContext(ctx)	8✔
245	if id == nil \|\| id.Tenant == "" {	8✔
NEW 246	// Not a tenant db - skip!	×
NEW 247	return nil	×
NEW 248	}	×
249	tenantId := id.Tenant	8✔
250	return tryPropagateReportingForTenant(db, wflows, tenantId, requestPeriod, dryRun)	8✔
251	}
252	if tenant != "" {	4✔
NEW 253	ctx := identity.WithContext(context.Background(),	×
NEW 254	&identity.Identity{	×
NEW 255	Tenant: tenant,	×
NEW 256	},	×
NEW 257	)	×
NEW 258	err := mapFunc(ctx)	×
UNCOV 259	if err != nil {	×
NEW 260	return errors.Wrap(err, "failed to propagate for given tenant")	×
NEW 261	}	×
NEW 262	l.Infof("tenant processed, exiting.")	×
263	} else {	4✔
264	err := db.ForEachTenant(context.Background(), mapFunc)	4✔
265	if err != nil {	4✔
NEW 266	return errors.Wrap(err, "failed to propagate for all tenant")	×
UNCOV 267	}	×
268	l.Info("all tenants processed, exiting.")	4✔
269	}
270	return nil	4✔
271	}
272
UNCOV 273	func selectDbs(db store.DataStore, tenant string) ([]string, error) {	×
UNCOV 274	l := log.NewEmpty()	×
UNCOV 275		×
UNCOV 276	var dbs []string	×
UNCOV 277		×
UNCOV 278	if tenant != "" {	×
UNCOV 279	l.Infof("propagating inventory for user-specified tenant %s", tenant)	×
UNCOV 280	n := mstore.DbNameForTenant(tenant, mongo.DbName)	×
UNCOV 281	dbs = []string{n}	×
UNCOV 282	}	×
283
UNCOV 284	return dbs, nil	×
285	}
286
287	const (
288	devicesBatchSize = 512
289	)
290
291	func updateDevicesStatus(
292	ctx context.Context,
293	db store.DataStore,
294	c cinv.Client,
295	tenant string,
296	status string,
297	dryRun bool,
298	) error {	55✔
299	var skip uint	55✔
300		55✔
301	skip = 0	55✔
302	for {	110✔
303	devices, err := db.GetDevices(ctx,	55✔
304	skip,	55✔
305	devicesBatchSize,	55✔
306	model.DeviceFilter{Status: []string{status}},	55✔
307	)	55✔
308	if err != nil {	70✔
309	return errors.Wrap(err, "failed to get devices")	15✔
310	}	15✔
311
312	if len(devices) < 1 {	40✔
313	break	×
314	}
315
316	deviceUpdates := make([]model.DeviceInventoryUpdate, len(devices))	40✔
317		40✔
318	for i, d := range devices {	120✔
319	deviceUpdates[i].Id = d.Id	80✔
320	deviceUpdates[i].Revision = d.Revision	80✔
321	}	80✔
322
323	if !dryRun {	80✔
324	err = c.SetDeviceStatus(ctx, tenant, deviceUpdates, status)	40✔
325	if err != nil {	55✔
326	return err	15✔
327	}	15✔
328	}
329
330	if len(devices) < devicesBatchSize {	50✔
331	break	25✔
332	} else {	×
333	skip += devicesBatchSize	×
334	}	×
335	}
336	return nil	25✔
337	}
338
339	func updateDevicesIdData(
340	ctx context.Context,
341	db store.DataStore,
342	c cinv.Client,
343	tenant string,
344	dryRun bool,
345	) error {	×
346	var skip uint	×
347		×
348	skip = 0	×
349	for {	×
350	devices, err := db.GetDevices(ctx, skip, devicesBatchSize, model.DeviceFilter{})	×
351	if err != nil {	×
352	return errors.Wrap(err, "failed to get devices")	×
353	}	×
354
355	if len(devices) < 1 {	×
356	break	×
357	}
358
359	if !dryRun {	×
360	for _, d := range devices {	×
361	err := c.SetDeviceIdentity(ctx, tenant, d.Id, d.IdDataStruct)	×
362	if err != nil {	×
363	return err	×
364	}	×
365	}
366	}
367
368	skip += devicesBatchSize	×
369	if len(devices) < devicesBatchSize {	×
370	break	×
371	}
372	}
373	return nil	×
374	}
375
376	func tryPropagateStatusesInventoryForTenant(
377	db store.DataStore,
378	c cinv.Client,
379	tenant string,
380	migrationVersion string,
381	dryRun bool,
382	) error {	11✔
383	l := log.NewEmpty()	11✔
384		11✔
385	l.Infof("propagating device statuses to inventory from tenant: %s", tenant)	11✔
386		11✔
387	ctx := context.Background()	11✔
388	if tenant != "" {	22✔
389	ctx = identity.WithContext(ctx, &identity.Identity{	11✔
390	Tenant: tenant,	11✔
391	})	11✔
392	}	11✔
393
394	var err error	11✔
395	var errReturned error	11✔
396	for _, status := range model.DevStatuses {	66✔
397	err = updateDevicesStatus(ctx, db, c, tenant, status, dryRun)	55✔
398	if err != nil {	85✔
399	l.Infof(	30✔
400	"Done with tenant %s status=%s, but there were errors: %s.",	30✔
401	tenant,	30✔
402	status,	30✔
403	err.Error(),	30✔
404	)	30✔
405	errReturned = err	30✔
406	} else {	55✔
407	l.Infof("Done with tenant %s status=%s", tenant, status)	25✔
408	}	25✔
409	}
410	if migrationVersion != "" && !dryRun {	13✔
411	if errReturned != nil {	2✔
412	l.Warnf(	×
NEW 413	"Will not store %s migration version for tenant %s due to errors.",	×
414	migrationVersion,	×
NEW 415	tenant,	×
416	)	×
417	} else {	2✔
418	version, err := migrate.NewVersion(migrationVersion)	2✔
419	if version == nil \|\| err != nil {	3✔
420	l.Warnf(	1✔
421	"Will not store %s migration version in %s.migration_info due to bad version"+	1✔
422	" provided.",	1✔
423	migrationVersion,	1✔
424	tenant,	1✔
425	)	1✔
426	errReturned = err	1✔
427	} else {	2✔
428	_ = db.StoreMigrationVersion(ctx, version)	1✔
429	}	1✔
430	}
431	}
432
433	return errReturned	11✔
434	}
435
436	func tryPropagateIdDataInventoryForDb(
437	db store.DataStore,
438	c cinv.Client,
439	dbname string,
440	dryRun bool,
441	) error {	×
442	l := log.NewEmpty()	×
443		×
444	l.Infof("propagating device id_data to inventory from DB: %s", dbname)	×
445		×
446	tenant := mstore.TenantFromDbName(dbname, mongo.DbName)	×
447		×
448	ctx := context.Background()	×
449	if tenant != "" {	×
450	ctx = identity.WithContext(ctx, &identity.Identity{	×
451	Tenant: tenant,	×
452	})	×
453	}	×
454
455	err := updateDevicesIdData(ctx, db, c, tenant, dryRun)	×
456	if err != nil {	×
457	l.Infof("Done with DB %s, but there were errors: %s.", dbname, err.Error())	×
458	} else {	×
459	l.Infof("Done with DB %s", dbname)	×
460	}	×
461
462	return err	×
463	}
464
465	func tryPropagateReportingForTenant(
466	db store.DataStore,
467	wflows orchestrator.ClientRunner,
468	tenant string,
469	requestPeriod time.Duration,
470	dryRun bool,
471	) error {	8✔
472	l := log.NewEmpty()	8✔
473		8✔
474	l.Infof("propagating device data to reporting for tenant %s", tenant)	8✔
475		8✔
476	ctx := context.Background()	8✔
477	if tenant != "" {	16✔
478	ctx = identity.WithContext(ctx, &identity.Identity{	8✔
479	Tenant: tenant,	8✔
480	})	8✔
481	} else {	8✔
NEW 482	return errors.New("you must provide a tenant id")	×
UNCOV 483	}	×
484
485	err := reindexDevicesReporting(ctx, requestPeriod, db, wflows, dryRun)	8✔
486	if err != nil {	10✔
487	l.Infof("Done with tenant %s, but there were errors: %s.", tenant, err.Error())	2✔
488	} else {	8✔
489	l.Infof("Done with tenant %s", tenant)	6✔
490	}	6✔
491
492	return err	8✔
493	}
494
495	func reindexDevicesReporting(
496	ctx context.Context,
497	requestPeriod time.Duration,
498	db store.DataStore,
499	wflows orchestrator.ClientRunner,
500	dryRun bool,
501	) error {	8✔
502	var skip uint	8✔
503		8✔
504	skip = 0	8✔
505	done := ctx.Done()	8✔
506	rateLimit := time.NewTicker(requestPeriod)	8✔
507	defer rateLimit.Stop()	8✔
508	for {	16✔
509	devices, err := db.GetDevices(ctx, skip, devicesBatchSize, model.DeviceFilter{})	8✔
510	if err != nil {	9✔
511	return errors.Wrap(err, "failed to get devices")	1✔
512	}	1✔
513
514	if len(devices) < 1 {	7✔
515	break	×
516	}
517
518	if !dryRun {	11✔
519	deviceIDs := make([]string, len(devices))	4✔
520	for i, d := range devices {	11✔
521	deviceIDs[i] = d.Id	7✔
522	}	7✔
523	err := wflows.SubmitReindexReportingBatch(ctx, deviceIDs)	4✔
524	if err != nil {	5✔
525	return err	1✔
526	}	1✔
527	}
528
529	skip += devicesBatchSize	6✔
530	if len(devices) < devicesBatchSize {	12✔
531	break	6✔
532	}
533	select {	×
534	case <-rateLimit.C:	×
535
536	case <-done:	×
537	return ctx.Err()	×
538	}
539	}
540	return nil	6✔
541	}
542
543	const (
544	WorkflowsDeviceLimitText = "@/etc/workflows-enterprise/data/device_limit_email.txt"
545	WorkflowsDeviceLimitHTML = "@/etc/workflows-enterprise/data/device_limit_email.html"
546	WorkflowsDeviceLimitSubject = "Device limit almost reached"
547	)
548
549	func warnTenantUsers(
550	ctx context.Context,
551	tenantID string,
552	tadm tenant.ClientRunner,
553	wflows orchestrator.ClientRunner,
554	remainingDevices uint,
555	) error {	6✔
556	users, err := tadm.GetTenantUsers(ctx, tenantID)	6✔
557	if err != nil {	7✔
558	// Log the event and continue with the other tenants	1✔
559	return err	1✔
560	}	1✔
561	for i := range users {	13✔
562	warnWFlow := orchestrator.DeviceLimitWarning{	8✔
563	RequestID: "deviceAuthAdmin",	8✔
564	RecipientEmail: users[i].Email,	8✔
565		8✔
566	Subject: WorkflowsDeviceLimitSubject,	8✔
567	Body: WorkflowsDeviceLimitText,	8✔
568	BodyHTML: WorkflowsDeviceLimitHTML,	8✔
569	RemainingDevices: &remainingDevices,	8✔
570	}	8✔
571	err = wflows.SubmitDeviceLimitWarning(ctx, warnWFlow)	8✔
572	if err != nil {	9✔
573	return err	1✔
574	}	1✔
575	}
576	return nil	4✔
577	}
578
579	// CheckDeviceLimits goes through all tenant databases and checks if the number
580	// of accepted devices is above a given threshold (in %) and sends an email
581	// to all registered users registered under the given tenant.
582	func CheckDeviceLimits(
583	threshold float64,
584	ds store.DataStore,
585	tadm tenant.ClientRunner,
586	wflows orchestrator.ClientRunner,
587	) error {	6✔
588	// Sanitize threshold	6✔
589	if threshold > 100.0 {	7✔
590	threshold = 100.0	1✔
591	} else if threshold < 0.0 {	7✔
592	threshold = 0.0	1✔
593	}	1✔
594	threshProportion := threshold / 100.0	6✔
595		6✔
596	// mapFunc is applied to all existing databases in datastore.	6✔
597	mapFunc := func(ctx context.Context) error {	19✔
598	id := identity.FromContext(ctx)	13✔
599	if id == nil \|\| id.Tenant == "" {	14✔
600	// Not a tenant db - skip!	1✔
601	return nil	1✔
602	}	1✔
603	tenantID := id.Tenant	12✔
604	l := log.FromContext(ctx)	12✔
605		12✔
606	lim, err := ds.GetLimit(ctx, model.LimitMaxDeviceCount)	12✔
607	if err != nil {	13✔
608	return err	1✔
609	}	1✔
610	n, err := ds.GetDevCountByStatus(ctx, model.DevStatusAccepted)	11✔
611	if err != nil {	12✔
612	return err	1✔
613	}	1✔
614	if float64(n) >= (float64(lim.Value) * threshProportion) {	16✔
615	// User is above limit	6✔
616		6✔
617	remainingUsers := uint(n) - uint(lim.Value)	6✔
618	err := warnTenantUsers(ctx, tenantID, tadm, wflows, remainingUsers)	6✔
619	if err != nil {	8✔
620	l.Warnf(`Failed to warn tenant "%s" `+	2✔
621	`users nearing device limit: %s`,	2✔
622	tenantID, err.Error(),	2✔
623	)	2✔
624	}	2✔
625	}
626	return nil	10✔
627	}
628	// Start looping through the databases.
629	return ds.ForEachTenant(	6✔
630	context.Background(),	6✔
631	mapFunc,	6✔
632	)	6✔
633	}

mendersoftware / deviceauth / 948596009

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