1495380963

Committed 14 Oct 2024 03:35PM UTC coverage: 70.373% (-2.5%) from 72.904%

Build # 1495380963

Build Type

Pull #101

gitlab-ci

Committed by

mineralsfree

Commit Message

feat: tenant list added

Ticket: MEN-7568
Changelog: None

Signed-off-by: Mikita Pilinka <mikita.pilinka@northern.tech>

Pull Request Pull Request #101: feat: tenant list added

Run Details

4406 of 6391 branches covered (68.94%)

Branch coverage included in aggregate %.

88 of 183 new or added lines in 10 files covered. (48.09%)

2623 existing lines in 65 files now uncovered.

36673 of 51982 relevant lines covered (70.55%)

31.07 hits per line

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

60.95

/backend/services/deviceauth/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/pkg/errors"

        "github.com/mendersoftware/mender-server/pkg/config"
        "github.com/mendersoftware/mender-server/pkg/identity"
        "github.com/mendersoftware/mender-server/pkg/log"
        "github.com/mendersoftware/mender-server/pkg/mongo/migrate"
        mstore "github.com/mendersoftware/mender-server/pkg/store"

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

var NowUnixMilis = utils.UnixMilis

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

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

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

}

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

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

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

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

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

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

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

        return nil
}

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

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

        return nil
}

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

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

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

        return nil
}

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

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

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

        return nil
}

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

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

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

        return nil
}

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

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

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

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

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

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

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

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

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

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

const (
        devicesBatchSize = 512
)

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

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

                if len(devices) < 1 {
                        break
                }

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

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

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

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

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

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

                if len(devices) < 1 {
                        break
                }

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

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

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

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

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

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

        return errReturned
}

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

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

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

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

        return err
}

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

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

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

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

        return err
}

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

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

                if len(devices) < 1 {
                        break
                }

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

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

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

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

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

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

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

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

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

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

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

mendersoftware / mender-server / 1495380963

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