1550729677

Committed 19 Nov 2024 11:39AM UTC coverage: 72.56% (-0.2%) from 72.771%

Build # 1550729677

Build Type

Pull #202

gitlab-ci

Committed by

alfrunes

Commit Message

test: Fix deviceauth.cache tests after changing client initialization

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

Pull Request Pull Request #202: :construction: MEN-7733: Rate limits for devices APIs

Run Details

4174 of 6072 branches covered (68.74%)

Branch coverage included in aggregate %.

135 of 380 new or added lines in 8 files covered. (35.53%)

86 existing lines in 4 files now uncovered.

42591 of 58378 relevant lines covered (72.96%)

15.45 hits per line

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

66.58

/backend/services/deviceauth/cache/cache.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 cache introduces API throttling based
// on redis, and functions for auth token management.
//
// Throttling mechanisms
//
// 1. Quota enforcement
//
// Based on https://redislabs.com/redis-best-practices/basic-rate-limiting/, but with a flexible
// interval (ratelimits.ApiQuota.IntervalSec).
// Current usage for a device lives under key:
//
// `tenant:<tid>:version<tenant_key_version>:device:<did>:quota:<interval_num>: <num_reqs>`
//
// expiring in the defined time window.
//
// 2. Burst control
//
// Implemented with a simple single key:
//
// `tenant:<tid>:version<tenant_key_version>:device:<did>:burst:<action>:<url>: <last_req_ts>`
//
// expiring in ratelimits.ApiBurst.MinIntervalSec.
// The value is not really important, just the existence of the key
// means the burst was exceeded.
//
// Token Management
//
// Tokens are expected at:
// `tenant:<tid>:version<tenant_key_version>:device:<did>:tok: <token>`
//
// Cache invalidation.
// We achive cache invalidation by incrementing tenant key version.
// Each tenant related key in the cache has to contain tenant key version.
// This way, by incrementing tenant key version, we invalidate all tenant
// related keys.

package cache

import (
        "context"
        "encoding/json"
        "fmt"
        "strconv"
        "time"

        "github.com/pkg/errors"
        "github.com/redis/go-redis/v9"

        "github.com/mendersoftware/mender-server/pkg/identity"
        "github.com/mendersoftware/mender-server/pkg/log"
        "github.com/mendersoftware/mender-server/pkg/ratelimits"

        "github.com/mendersoftware/mender-server/services/deviceauth/model"
        "github.com/mendersoftware/mender-server/services/deviceauth/utils"
)

const (
        IdTypeDevice = "device"
        IdTypeUser   = "user"
        // expiration of the device check in time - one week
        CheckInTimeExpiration = time.Duration(time.Hour * 24 * 7)
)

var (
        ErrNoPositiveInteger = errors.New("must be a positive integer")
        ErrNegativeInteger   = errors.New("cannot be a negative integer")

        ErrTooManyRequests = errors.New("too many requests")
)

//go:generate ../../../utils/mockgen.sh
type Cache interface {
        // Throttle applies desired api limits and retrieves a cached token.
        // These ops are bundled because the implementation will pipeline them for a single network
        // roundtrip for max performance.
        // Returns:
        // - the token (if any)
        // - potentially ErrTooManyRequests (other errors: internal)
        Throttle(
                ctx context.Context,
                rawToken string,
                l ratelimits.ApiLimits,
                tid,
                id,
                idtype,
                url,
                action string,
        ) (string, error)

        // CacheToken caches the token under designated key, with expiration
        CacheToken(ctx context.Context, tid, id, idtype, token string, expireSec time.Duration) error

        // DeleteToken deletes the token for 'id'
        DeleteToken(ctx context.Context, tid, id, idtype string) error

        // GetLimit gets a limit from cache (see store.Datastore.GetLimit)
        GetLimit(ctx context.Context, name string) (*model.Limit, error)
        // SetLimit writes a limit to cache (see store.Datastore.SetLimit)
        SetLimit(ctx context.Context, limit *model.Limit) error
        // DeleteLimit evicts the limit with the given name from cache
        DeleteLimit(ctx context.Context, name string) error

        // GetLimits fetches limits for 'id'
        GetLimits(ctx context.Context, tid, id, idtype string) (*ratelimits.ApiLimits, error)

        // CacheLimits saves limits for 'id'
        CacheLimits(ctx context.Context, l ratelimits.ApiLimits, tid, id, idtype string) error

        // CacheCheckInTime caches the last device check in time
        CacheCheckInTime(ctx context.Context, t *time.Time, tid, id string) error

        // GetCheckInTime gets the last device check in time from cache
        GetCheckInTime(ctx context.Context, tid, id string) (*time.Time, error)

        // GetCheckInTimes gets the last device check in time from cache
        // for each device with id from the list of ids
        GetCheckInTimes(ctx context.Context, tid string, ids []string) ([]*time.Time, error)

        // SuspendTenant increment tenant key version
        // tenant key is used in all cache keys, this way, when we increment the key version,
        // all the keys are no longer accessible - in other words, be incrementing tenant key version
        // we invalidate all tenant keys
        SuspendTenant(ctx context.Context, tid string) error
}

type RedisCache struct {
        c               redis.Cmdable
        prefix          string
        LimitsExpireSec int
        DefaultExpire   time.Duration
        clock           utils.Clock
}

func NewRedisCache(
        redisClient redis.Cmdable,
        prefix string,
        limitsExpireSec int,
) *RedisCache {
        return &RedisCache{
                c:               redisClient,
                LimitsExpireSec: limitsExpireSec,
                prefix:          prefix,
                DefaultExpire:   time.Hour * 3,
                clock:           utils.NewClock(),
        }
}

func (rl *RedisCache) WithClock(c utils.Clock) *RedisCache {
        rl.clock = c
        return rl
}

func (rl *RedisCache) keyLimit(tenantID, name string) string {
        if tenantID == "" {
                tenantID = "default"
        }
        return fmt.Sprintf("%s:tenant:%s:limit:%s", rl.prefix, tenantID, name)
}

func (rl *RedisCache) GetLimit(ctx context.Context, name string) (*model.Limit, error) {
        var tenantID string
        id := identity.FromContext(ctx)
        if id != nil {
                tenantID = id.Tenant
        }
        value, err := rl.c.Get(ctx, rl.keyLimit(tenantID, name)).Uint64()
        if err != nil {
                if errors.Is(err, redis.Nil) {
                        return nil, nil
                }
                return nil, err
        }
        return &model.Limit{
                TenantID: tenantID,
                Value:    value,
                Name:     name,
        }, nil
}

func (rl *RedisCache) SetLimit(ctx context.Context, limit *model.Limit) error {
        if limit == nil {
                return nil
        }
        var tenantID string
        id := identity.FromContext(ctx)
        if id != nil {
                tenantID = id.Tenant
        }
        key := rl.keyLimit(tenantID, limit.Name)
        return rl.c.SetEx(ctx, key, limit.Value, rl.DefaultExpire).Err()
}

func (rl *RedisCache) DeleteLimit(ctx context.Context, name string) error {
        var tenantID string
        id := identity.FromContext(ctx)
        if id != nil {
                tenantID = id.Tenant
        }
        key := rl.keyLimit(tenantID, name)
        return rl.c.Del(ctx, key).Err()
}

func (rl *RedisCache) Throttle(
        ctx context.Context,
        rawToken string,
        l ratelimits.ApiLimits,
        tid,
        id,
        idtype,
        url,
        action string,
) (string, error) {
        now := rl.clock.Now().Unix()

        var tokenGet *redis.StringCmd
        var quotaInc *redis.IntCmd
        var quotaExp *redis.BoolCmd
        var burstGet *redis.StringCmd
        var burstSet *redis.StatusCmd

        pipe := rl.c.Pipeline()

        version, err := rl.getTenantKeyVersion(ctx, tid)
        if err != nil {
                return "", err
        }

        // queue quota/burst control and token fetching
        // for piped execution
        quotaInc, quotaExp = rl.pipeQuota(ctx, pipe, l, tid, id, idtype, now, version)
        tokenGet = rl.pipeToken(ctx, pipe, tid, id, idtype, version)

        burstGet, burstSet = rl.pipeBurst(ctx,
                pipe,
                l,
                tid, id, idtype,
                url, action,
                now, version)

        _, err = pipe.Exec(ctx)
        if err != nil && !isErrRedisNil(err) {
                return "", err
        }

        // collect quota/burst control and token fetch results
        tok, err := rl.checkToken(tokenGet, rawToken)
        if err != nil {
                return "", err
        }

        err = rl.checkQuota(l, quotaInc, quotaExp)
        if err != nil {
                return "", err
        }

        err = rl.checkBurst(burstGet, burstSet)
        if err != nil {
                return "", err
        }

        return tok, nil
}

func (rl *RedisCache) pipeToken(
        ctx context.Context,
        pipe redis.Pipeliner,
        tid,
        id,
        idtype string,
        version int,
) *redis.StringCmd {
        key := rl.KeyToken(tid, id, idtype, version)
        return pipe.Get(ctx, key)
}

func (rl *RedisCache) checkToken(cmd *redis.StringCmd, raw string) (string, error) {
        err := cmd.Err()
        if err != nil {
                if isErrRedisNil(err) {
                        return "", nil
                }
                return "", err
        }

        token := cmd.Val()
        if token == raw {
                return token, nil
        } else {
                // must be a stale token - we don't want to use it
                // let it expire in the background
                return "", nil
        }
}

func (rl *RedisCache) pipeQuota(
        ctx context.Context,
        pipe redis.Pipeliner,
        l ratelimits.ApiLimits,
        tid,
        id,
        idtype string,
        now int64,
        version int,
) (*redis.IntCmd, *redis.BoolCmd) {
        var incr *redis.IntCmd
        var expire *redis.BoolCmd

        // not a default/empty quota
        if l.ApiQuota.MaxCalls != 0 {
                intvl := int64(now / int64(l.ApiQuota.IntervalSec))
                keyQuota := rl.KeyQuota(tid, id, idtype, strconv.FormatInt(intvl, 10), version)
                incr = pipe.Incr(ctx, keyQuota)
                expire = pipe.Expire(ctx, keyQuota, time.Duration(l.ApiQuota.IntervalSec)*time.Second)
        }

        return incr, expire
}

func (rl *RedisCache) checkQuota(
        l ratelimits.ApiLimits,
        incr *redis.IntCmd,
        expire *redis.BoolCmd,
) error {
        if incr == nil && expire == nil {
                return nil
        }

        err := incr.Err()
        if err != nil && !isErrRedisNil(err) {
                return err
        }

        err = expire.Err()
        if err != nil {
                return err
        }

        quota := incr.Val()
        if quota > int64(l.ApiQuota.MaxCalls) {
                return ErrTooManyRequests
        }

        return nil
}

func (rl *RedisCache) pipeBurst(ctx context.Context,
        pipe redis.Pipeliner,
        l ratelimits.ApiLimits,
        tid, id, idtype, url, action string,
        now int64, version int) (*redis.StringCmd, *redis.StatusCmd) {
        var get *redis.StringCmd
        var set *redis.StatusCmd

        for _, b := range l.ApiBursts {
                if b.Action == action &&
                        b.Uri == url &&
                        b.MinIntervalSec != 0 {

                        intvl := int64(now / int64(b.MinIntervalSec))
                        keyBurst := rl.KeyBurst(
                                tid, id, idtype, url, action, strconv.FormatInt(intvl, 10), version)

                        get = pipe.Get(ctx, keyBurst)
                        set = pipe.Set(ctx, keyBurst, now, time.Duration(b.MinIntervalSec)*time.Second)
                }
        }

        return get, set
}

func (rl *RedisCache) checkBurst(get *redis.StringCmd, set *redis.StatusCmd) error {
        if get != nil && set != nil {
                err := get.Err()

                // no error means burst was found/hit
                if err == nil {
                        return ErrTooManyRequests
                }

                if isErrRedisNil(err) {
                        return nil
                }

                return err
        }

        return nil
}

func (rl *RedisCache) CacheToken(
        ctx context.Context,
        tid,
        id,
        idtype,
        token string,
        expire time.Duration,
) error {
        version, err := rl.getTenantKeyVersion(ctx, tid)
        if err != nil {
                return err
        }
        res := rl.c.Set(ctx, rl.KeyToken(tid, id, idtype, version),
                token,
                expire)
        return res.Err()
}

func (rl *RedisCache) DeleteToken(ctx context.Context, tid, id, idtype string) error {
        version, err := rl.getTenantKeyVersion(ctx, tid)
        if err != nil {
                return err
        }
        res := rl.c.Del(ctx, rl.KeyToken(tid, id, idtype, version))
        return res.Err()
}

func (rl *RedisCache) GetLimits(
        ctx context.Context,
        tid,
        id,
        idtype string,
) (*ratelimits.ApiLimits, error) {
        version, err := rl.getTenantKeyVersion(ctx, tid)
        if err != nil {
                return nil, err
        }

        res := rl.c.Get(ctx, rl.KeyLimits(tid, id, idtype, version))

        if res.Err() != nil {
                if isErrRedisNil(res.Err()) {
                        return nil, nil
                }
                return nil, res.Err()
        }

        var limits ratelimits.ApiLimits

        err = json.Unmarshal([]byte(res.Val()), &limits)
        if err != nil {
                return nil, err
        }

        return &limits, nil
}

func (rl *RedisCache) CacheLimits(
        ctx context.Context,
        l ratelimits.ApiLimits,
        tid,
        id,
        idtype string,
) error {
        enc, err := json.Marshal(l)
        if err != nil {
                return err
        }

        version, err := rl.getTenantKeyVersion(ctx, tid)
        if err != nil {
                return err
        }

        res := rl.c.Set(
                ctx,
                rl.KeyLimits(tid, id, idtype, version),
                enc,
                time.Duration(rl.LimitsExpireSec)*time.Second,
        )

        return res.Err()
}

func (rl *RedisCache) KeyQuota(tid, id, idtype, intvlNum string, version int) string {
        return fmt.Sprintf(
                "%s:tenant:%s:version:%d:%s:%s:quota:%s",
                rl.prefix, tid, version, idtype, id, intvlNum)
}

func (rl *RedisCache) KeyBurst(
        tid, id, idtype, url, action, intvlNum string, version int,
) string {
        return fmt.Sprintf(
                "%s:tenant:%s:version:%d:%s:%s:burst:%s:%s:%s",
                rl.prefix, tid, version, idtype, id, url, action, intvlNum)
}

func (rl *RedisCache) KeyToken(tid, id, idtype string, version int) string {
        return fmt.Sprintf(
                "%s:tenant:%s:version:%d:%s:%s:tok",
                rl.prefix, tid, version, idtype, id)
}

func (rl *RedisCache) KeyLimits(tid, id, idtype string, version int) string {
        return fmt.Sprintf(
                "%s:tenant:%s:version:%d:%s:%s:limits",
                rl.prefix, tid, version, idtype, id)
}

func (rl *RedisCache) KeyCheckInTime(tid, id, idtype string, version int) string {
        return fmt.Sprintf(
                "%s:tenant:%s:version:%d:%s:%s:checkInTime",
                rl.prefix, tid, version, idtype, id)
}

func (rl *RedisCache) KeyTenantVersion(tid string) string {
        return fmt.Sprintf("%s:tenant:%s:version", rl.prefix, tid)
}

// isErrRedisNil checks for a very common non-error, "redis: nil",
// which just means the key was not found, and is normal
// it's routinely returned e.g. from GET, or pipelines containing it
func isErrRedisNil(e error) bool {
        return errors.Is(e, redis.Nil)
}

// TODO: move to go-lib-micro/ratelimits
func LimitsEmpty(l *ratelimits.ApiLimits) bool {
        return l.ApiQuota.MaxCalls == 0 &&
                l.ApiQuota.IntervalSec == 0 &&
                len(l.ApiBursts) == 0
}

func (rl *RedisCache) CacheCheckInTime(
        ctx context.Context,
        t *time.Time,
        tid,
        id string,
) error {
        tj, err := json.Marshal(t)
        if err != nil {
                return err
        }

        version, err := rl.getTenantKeyVersion(ctx, tid)
        if err != nil {
                return err
        }

        res := rl.c.Set(
                ctx,
                rl.KeyCheckInTime(tid, id, IdTypeDevice, version),
                tj,
                CheckInTimeExpiration,
        )

        return res.Err()
}

func (rl *RedisCache) GetCheckInTime(
        ctx context.Context,
        tid,
        id string,
) (*time.Time, error) {
        version, err := rl.getTenantKeyVersion(ctx, tid)
        if err != nil {
                return nil, err
        }

        res := rl.c.Get(ctx, rl.KeyCheckInTime(tid, id, IdTypeDevice, version))

        if res.Err() != nil {
                if isErrRedisNil(res.Err()) {
                        return nil, nil
                }
                return nil, res.Err()
        }

        var checkInTime time.Time

        err = json.Unmarshal([]byte(res.Val()), &checkInTime)
        if err != nil {
                return nil, err
        }

        return &checkInTime, nil
}

func (rl *RedisCache) GetCheckInTimes(
        ctx context.Context,
        tid string,
        ids []string,
) ([]*time.Time, error) {
        l := log.FromContext(ctx)

        version, err := rl.getTenantKeyVersion(ctx, tid)
        if err != nil {
                return nil, err
        }
        checkInTimes := make([]*time.Time, len(ids))
        if _, ok := rl.c.(*redis.ClusterClient); ok {
                pipe := rl.c.Pipeline()
                for _, id := range ids {
                        pipe.Get(ctx, rl.KeyCheckInTime(tid, id, IdTypeDevice, version))
                }
                results, err := pipe.Exec(ctx)
                if err != nil && !errors.Is(err, redis.Nil) {
                        return nil, fmt.Errorf("failed to fetch check in times: %w", err)
                }
                for i, result := range results {
                        cmd, ok := result.(*redis.StringCmd)
                        if !ok {
                                continue // should never happen
                        }
                        b, err := cmd.Bytes()
                        if err != nil {
                                if errors.Is(err, redis.Nil) {
                                        continue
                                } else {
                                        l.Errorf("failed to get device: %s", err.Error())
                                }
                        } else {
                                checkInTime := new(time.Time)
                                err = json.Unmarshal(b, checkInTime)
                                if err != nil {
                                        l.Errorf("failed to deserialize check in time: %s", err.Error())
                                } else {
                                        checkInTimes[i] = checkInTime
                                }

                        }
                }
        } else {
                keys := make([]string, len(ids))
                for i, id := range ids {
                        keys[i] = rl.KeyCheckInTime(tid, id, IdTypeDevice, version)
                }
                res := rl.c.MGet(ctx, keys...)

                for i, v := range res.Val() {
                        if v != nil {
                                b, ok := v.(string)
                                if !ok {
                                        continue
                                }
                                var checkInTime time.Time
                                err := json.Unmarshal([]byte(b), &checkInTime)
                                if err != nil {
                                        l.Errorf("failed to unmarshal check-in time: %s", err.Error())
                                        continue
                                }
                                checkInTimes[i] = &checkInTime
                        }
                }
        }

        return checkInTimes, nil
}

func (rl *RedisCache) SuspendTenant(
        ctx context.Context,
        tid string,
) error {
        res := rl.c.Incr(ctx, rl.KeyTenantVersion(tid))
        return res.Err()
}

func (rl *RedisCache) getTenantKeyVersion(ctx context.Context, tid string) (int, error) {
        res := rl.c.Get(ctx, rl.KeyTenantVersion(tid))
        if res.Err() != nil {
                if isErrRedisNil(res.Err()) {
                        return 0, nil
                }
                return 0, res.Err()
        }

        var version int

        err := json.Unmarshal([]byte(res.Val()), &version)
        if err != nil {
                return 0, err
        }

        return version, nil
}

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
15	// Package cache introduces API throttling based
16	// on redis, and functions for auth token management.
17	//
18	// Throttling mechanisms
19	//
20	// 1. Quota enforcement
21	//
22	// Based on https://redislabs.com/redis-best-practices/basic-rate-limiting/, but with a flexible
23	// interval (ratelimits.ApiQuota.IntervalSec).
24	// Current usage for a device lives under key:
25	//
26	// `tenant:<tid>:version<tenant_key_version>:device:<did>:quota:<interval_num>: <num_reqs>`
27	//
28	// expiring in the defined time window.
29	//
30	// 2. Burst control
31	//
32	// Implemented with a simple single key:
33	//
34	// `tenant:<tid>:version<tenant_key_version>:device:<did>:burst:<action>:<url>: <last_req_ts>`
35	//
36	// expiring in ratelimits.ApiBurst.MinIntervalSec.
37	// The value is not really important, just the existence of the key
38	// means the burst was exceeded.
39	//
40	// Token Management
41	//
42	// Tokens are expected at:
43	// `tenant:<tid>:version<tenant_key_version>:device:<did>:tok: <token>`
44	//
45	// Cache invalidation.
46	// We achive cache invalidation by incrementing tenant key version.
47	// Each tenant related key in the cache has to contain tenant key version.
48	// This way, by incrementing tenant key version, we invalidate all tenant
49	// related keys.
50
51	package cache
52
53	import (
54	"context"
55	"encoding/json"
56	"fmt"
57	"strconv"
58	"time"
59
60	"github.com/pkg/errors"
61	"github.com/redis/go-redis/v9"
62
63	"github.com/mendersoftware/mender-server/pkg/identity"
64	"github.com/mendersoftware/mender-server/pkg/log"
65	"github.com/mendersoftware/mender-server/pkg/ratelimits"
66
67	"github.com/mendersoftware/mender-server/services/deviceauth/model"
68	"github.com/mendersoftware/mender-server/services/deviceauth/utils"
69	)
70
71	const (
72	IdTypeDevice = "device"
73	IdTypeUser = "user"
74	// expiration of the device check in time - one week
75	CheckInTimeExpiration = time.Duration(time.Hour * 24 * 7)
76	)
77
78	var (
79	ErrNoPositiveInteger = errors.New("must be a positive integer")
80	ErrNegativeInteger = errors.New("cannot be a negative integer")
81
82	ErrTooManyRequests = errors.New("too many requests")
83	)
84
85	//go:generate ../../../utils/mockgen.sh
86	type Cache interface {
87	// Throttle applies desired api limits and retrieves a cached token.
88	// These ops are bundled because the implementation will pipeline them for a single network
89	// roundtrip for max performance.
90	// Returns:
91	// - the token (if any)
92	// - potentially ErrTooManyRequests (other errors: internal)
93	Throttle(
94	ctx context.Context,
95	rawToken string,
96	l ratelimits.ApiLimits,
97	tid,
98	id,
99	idtype,
100	url,
101	action string,
102	) (string, error)
103
104	// CacheToken caches the token under designated key, with expiration
105	CacheToken(ctx context.Context, tid, id, idtype, token string, expireSec time.Duration) error
106
107	// DeleteToken deletes the token for 'id'
108	DeleteToken(ctx context.Context, tid, id, idtype string) error
109
110	// GetLimit gets a limit from cache (see store.Datastore.GetLimit)
111	GetLimit(ctx context.Context, name string) (*model.Limit, error)
112	// SetLimit writes a limit to cache (see store.Datastore.SetLimit)
113	SetLimit(ctx context.Context, limit *model.Limit) error
114	// DeleteLimit evicts the limit with the given name from cache
115	DeleteLimit(ctx context.Context, name string) error
116
117	// GetLimits fetches limits for 'id'
118	GetLimits(ctx context.Context, tid, id, idtype string) (*ratelimits.ApiLimits, error)
119
120	// CacheLimits saves limits for 'id'
121	CacheLimits(ctx context.Context, l ratelimits.ApiLimits, tid, id, idtype string) error
122
123	// CacheCheckInTime caches the last device check in time
124	CacheCheckInTime(ctx context.Context, t *time.Time, tid, id string) error
125
126	// GetCheckInTime gets the last device check in time from cache
127	GetCheckInTime(ctx context.Context, tid, id string) (*time.Time, error)
128
129	// GetCheckInTimes gets the last device check in time from cache
130	// for each device with id from the list of ids
131	GetCheckInTimes(ctx context.Context, tid string, ids []string) ([]*time.Time, error)
132
133	// SuspendTenant increment tenant key version
134	// tenant key is used in all cache keys, this way, when we increment the key version,
135	// all the keys are no longer accessible - in other words, be incrementing tenant key version
136	// we invalidate all tenant keys
137	SuspendTenant(ctx context.Context, tid string) error
138	}
139
140	type RedisCache struct {
141	c redis.Cmdable
142	prefix string
143	LimitsExpireSec int
144	DefaultExpire time.Duration
145	clock utils.Clock
146	}
147
148	func NewRedisCache(
149	redisClient redis.Cmdable,
150	prefix string,
151	limitsExpireSec int,
152	) *RedisCache {	1✔
153	return &RedisCache{	1✔
154	c: redisClient,	1✔
155	LimitsExpireSec: limitsExpireSec,	1✔
156	prefix: prefix,	1✔
157	DefaultExpire: time.Hour * 3,	1✔
158	clock: utils.NewClock(),	1✔
159	}	1✔
160	}	1✔
161
162	func (rl RedisCache) WithClock(c utils.Clock) RedisCache {	1✔
163	rl.clock = c	1✔
164	return rl	1✔
165	}	1✔
166
NEW 167	func (rl *RedisCache) keyLimit(tenantID, name string) string {	×
NEW 168	if tenantID == "" {	×
NEW 169	tenantID = "default"	×
NEW 170	}	×
NEW 171	return fmt.Sprintf("%s:tenant:%s:limit:%s", rl.prefix, tenantID, name)	×
172	}
173
NEW 174	func (rl RedisCache) GetLimit(ctx context.Context, name string) (model.Limit, error) {	×
NEW 175	var tenantID string	×
NEW 176	id := identity.FromContext(ctx)	×
NEW 177	if id != nil {	×
NEW 178	tenantID = id.Tenant	×
NEW 179	}	×
NEW 180	value, err := rl.c.Get(ctx, rl.keyLimit(tenantID, name)).Uint64()	×
NEW 181	if err != nil {	×
NEW 182	if errors.Is(err, redis.Nil) {	×
NEW 183	return nil, nil	×
NEW 184	}	×
NEW 185	return nil, err	×
186	}
NEW 187	return &model.Limit{	×
NEW 188	TenantID: tenantID,	×
NEW 189	Value: value,	×
NEW 190	Name: name,	×
NEW 191	}, nil	×
192	}
193
NEW 194	func (rl RedisCache) SetLimit(ctx context.Context, limit model.Limit) error {	×
NEW 195	if limit == nil {	×
NEW 196	return nil	×
NEW 197	}	×
NEW 198	var tenantID string	×
NEW 199	id := identity.FromContext(ctx)	×
NEW 200	if id != nil {	×
NEW 201	tenantID = id.Tenant	×
NEW 202	}	×
NEW 203	key := rl.keyLimit(tenantID, limit.Name)	×
NEW 204	return rl.c.SetEx(ctx, key, limit.Value, rl.DefaultExpire).Err()	×
205	}
206
NEW 207	func (rl *RedisCache) DeleteLimit(ctx context.Context, name string) error {	×
NEW 208	var tenantID string	×
NEW 209	id := identity.FromContext(ctx)	×
NEW 210	if id != nil {	×
NEW 211	tenantID = id.Tenant	×
NEW 212	}	×
NEW 213	key := rl.keyLimit(tenantID, name)	×
NEW 214	return rl.c.Del(ctx, key).Err()	×
215	}
216
217	func (rl *RedisCache) Throttle(
218	ctx context.Context,
219	rawToken string,
220	l ratelimits.ApiLimits,
221	tid,
222	id,
223	idtype,
224	url,
225	action string,
226	) (string, error) {	1✔
227	now := rl.clock.Now().Unix()	1✔
228		1✔
229	var tokenGet *redis.StringCmd	1✔
230	var quotaInc *redis.IntCmd	1✔
231	var quotaExp *redis.BoolCmd	1✔
232	var burstGet *redis.StringCmd	1✔
233	var burstSet *redis.StatusCmd	1✔
234		1✔
235	pipe := rl.c.Pipeline()	1✔
236		1✔
237	version, err := rl.getTenantKeyVersion(ctx, tid)	1✔
238	if err != nil {	1✔
239	return "", err	×
240	}	×
241
242	// queue quota/burst control and token fetching
243	// for piped execution
244	quotaInc, quotaExp = rl.pipeQuota(ctx, pipe, l, tid, id, idtype, now, version)	1✔
245	tokenGet = rl.pipeToken(ctx, pipe, tid, id, idtype, version)	1✔
246		1✔
247	burstGet, burstSet = rl.pipeBurst(ctx,	1✔
248	pipe,	1✔
249	l,	1✔
250	tid, id, idtype,	1✔
251	url, action,	1✔
252	now, version)	1✔
253		1✔
254	_, err = pipe.Exec(ctx)	1✔
255	if err != nil && !isErrRedisNil(err) {	1✔
256	return "", err	×
257	}	×
258
259	// collect quota/burst control and token fetch results
260	tok, err := rl.checkToken(tokenGet, rawToken)	1✔
261	if err != nil {	1✔
262	return "", err	×
263	}	×
264
265	err = rl.checkQuota(l, quotaInc, quotaExp)	1✔
266	if err != nil {	2✔
267	return "", err	1✔
268	}	1✔
269
270	err = rl.checkBurst(burstGet, burstSet)	1✔
271	if err != nil {	2✔
272	return "", err	1✔
273	}	1✔
274
275	return tok, nil	1✔
276	}
277
278	func (rl *RedisCache) pipeToken(
279	ctx context.Context,
280	pipe redis.Pipeliner,
281	tid,
282	id,
283	idtype string,
284	version int,
285	) *redis.StringCmd {	1✔
286	key := rl.KeyToken(tid, id, idtype, version)	1✔
287	return pipe.Get(ctx, key)	1✔
288	}	1✔
289
290	func (rl RedisCache) checkToken(cmd redis.StringCmd, raw string) (string, error) {	1✔
291	err := cmd.Err()	1✔
292	if err != nil {	2✔
293	if isErrRedisNil(err) {	2✔
294	return "", nil	1✔
295	}	1✔
296	return "", err	×
297	}
298
299	token := cmd.Val()	1✔
300	if token == raw {	2✔
301	return token, nil	1✔
302	} else {	2✔
303	// must be a stale token - we don't want to use it	1✔
304	// let it expire in the background	1✔
305	return "", nil	1✔
306	}	1✔
307	}
308
309	func (rl *RedisCache) pipeQuota(
310	ctx context.Context,
311	pipe redis.Pipeliner,
312	l ratelimits.ApiLimits,
313	tid,
314	id,
315	idtype string,
316	now int64,
317	version int,
318	) (redis.IntCmd, redis.BoolCmd) {	1✔
319	var incr *redis.IntCmd	1✔
320	var expire *redis.BoolCmd	1✔
321		1✔
322	// not a default/empty quota	1✔
323	if l.ApiQuota.MaxCalls != 0 {	2✔
324	intvl := int64(now / int64(l.ApiQuota.IntervalSec))	1✔
325	keyQuota := rl.KeyQuota(tid, id, idtype, strconv.FormatInt(intvl, 10), version)	1✔
326	incr = pipe.Incr(ctx, keyQuota)	1✔
327	expire = pipe.Expire(ctx, keyQuota, time.Duration(l.ApiQuota.IntervalSec)*time.Second)	1✔
328	}	1✔
329
330	return incr, expire	1✔
331	}
332
333	func (rl *RedisCache) checkQuota(
334	l ratelimits.ApiLimits,
335	incr *redis.IntCmd,
336	expire *redis.BoolCmd,
337	) error {	1✔
338	if incr == nil && expire == nil {	2✔
339	return nil	1✔
340	}	1✔
341
342	err := incr.Err()	1✔
343	if err != nil && !isErrRedisNil(err) {	1✔
344	return err	×
345	}	×
346
347	err = expire.Err()	1✔
348	if err != nil {	1✔
349	return err	×
350	}	×
351
352	quota := incr.Val()	1✔
353	if quota > int64(l.ApiQuota.MaxCalls) {	2✔
354	return ErrTooManyRequests	1✔
355	}	1✔
356
357	return nil	1✔
358	}
359
360	func (rl *RedisCache) pipeBurst(ctx context.Context,
361	pipe redis.Pipeliner,
362	l ratelimits.ApiLimits,
363	tid, id, idtype, url, action string,
364	now int64, version int) (redis.StringCmd, redis.StatusCmd) {	1✔
365	var get *redis.StringCmd	1✔
366	var set *redis.StatusCmd	1✔
367		1✔
368	for _, b := range l.ApiBursts {	2✔
369	if b.Action == action &&	1✔
370	b.Uri == url &&	1✔
371	b.MinIntervalSec != 0 {	2✔
372		1✔
373	intvl := int64(now / int64(b.MinIntervalSec))	1✔
374	keyBurst := rl.KeyBurst(	1✔
375	tid, id, idtype, url, action, strconv.FormatInt(intvl, 10), version)	1✔
376		1✔
377	get = pipe.Get(ctx, keyBurst)	1✔
378	set = pipe.Set(ctx, keyBurst, now, time.Duration(b.MinIntervalSec)*time.Second)	1✔
379	}	1✔
380	}
381
382	return get, set	1✔
383	}
384
385	func (rl RedisCache) checkBurst(get redis.StringCmd, set *redis.StatusCmd) error {	1✔
386	if get != nil && set != nil {	2✔
387	err := get.Err()	1✔
388		1✔
389	// no error means burst was found/hit	1✔
390	if err == nil {	2✔
391	return ErrTooManyRequests	1✔
392	}	1✔
393
394	if isErrRedisNil(err) {	2✔
395	return nil	1✔
396	}	1✔
397
398	return err	×
399	}
400
401	return nil	1✔
402	}
403
404	func (rl *RedisCache) CacheToken(
405	ctx context.Context,
406	tid,
407	id,
408	idtype,
409	token string,
410	expire time.Duration,
411	) error {	1✔
412	version, err := rl.getTenantKeyVersion(ctx, tid)	1✔
413	if err != nil {	1✔
414	return err	×
415	}	×
416	res := rl.c.Set(ctx, rl.KeyToken(tid, id, idtype, version),	1✔
417	token,	1✔
418	expire)	1✔
419	return res.Err()	1✔
420	}
421
422	func (rl *RedisCache) DeleteToken(ctx context.Context, tid, id, idtype string) error {	1✔
423	version, err := rl.getTenantKeyVersion(ctx, tid)	1✔
424	if err != nil {	1✔
425	return err	×
426	}	×
427	res := rl.c.Del(ctx, rl.KeyToken(tid, id, idtype, version))	1✔
428	return res.Err()	1✔
429	}
430
431	func (rl *RedisCache) GetLimits(
432	ctx context.Context,
433	tid,
434	id,
435	idtype string,
436	) (*ratelimits.ApiLimits, error) {	1✔
437	version, err := rl.getTenantKeyVersion(ctx, tid)	1✔
438	if err != nil {	1✔
439	return nil, err	×
440	}	×
441
442	res := rl.c.Get(ctx, rl.KeyLimits(tid, id, idtype, version))	1✔
443		1✔
444	if res.Err() != nil {	2✔
445	if isErrRedisNil(res.Err()) {	2✔
446	return nil, nil	1✔
447	}	1✔
448	return nil, res.Err()	×
449	}
450
451	var limits ratelimits.ApiLimits	1✔
452		1✔
453	err = json.Unmarshal([]byte(res.Val()), &limits)	1✔
454	if err != nil {	1✔
455	return nil, err	×
456	}	×
457
458	return &limits, nil	1✔
459	}
460
461	func (rl *RedisCache) CacheLimits(
462	ctx context.Context,
463	l ratelimits.ApiLimits,
464	tid,
465	id,
466	idtype string,
467	) error {	1✔
468	enc, err := json.Marshal(l)	1✔
469	if err != nil {	1✔
470	return err	×
471	}	×
472
473	version, err := rl.getTenantKeyVersion(ctx, tid)	1✔
474	if err != nil {	1✔
475	return err	×
476	}	×
477
478	res := rl.c.Set(	1✔
479	ctx,	1✔
480	rl.KeyLimits(tid, id, idtype, version),	1✔
481	enc,	1✔
482	time.Duration(rl.LimitsExpireSec)*time.Second,	1✔
483	)	1✔
484		1✔
485	return res.Err()	1✔
486	}
487
488	func (rl *RedisCache) KeyQuota(tid, id, idtype, intvlNum string, version int) string {	1✔
489	return fmt.Sprintf(	1✔
490	"%s:tenant:%s:version:%d:%s:%s:quota:%s",	1✔
491	rl.prefix, tid, version, idtype, id, intvlNum)	1✔
492	}	1✔
493
494	func (rl *RedisCache) KeyBurst(
495	tid, id, idtype, url, action, intvlNum string, version int,
496	) string {	1✔
497	return fmt.Sprintf(	1✔
498	"%s:tenant:%s:version:%d:%s:%s:burst:%s:%s:%s",	1✔
499	rl.prefix, tid, version, idtype, id, url, action, intvlNum)	1✔
500	}	1✔
501
502	func (rl *RedisCache) KeyToken(tid, id, idtype string, version int) string {	1✔
503	return fmt.Sprintf(	1✔
504	"%s:tenant:%s:version:%d:%s:%s:tok",	1✔
505	rl.prefix, tid, version, idtype, id)	1✔
506	}	1✔
507
508	func (rl *RedisCache) KeyLimits(tid, id, idtype string, version int) string {	1✔
509	return fmt.Sprintf(	1✔
510	"%s:tenant:%s:version:%d:%s:%s:limits",	1✔
511	rl.prefix, tid, version, idtype, id)	1✔
512	}	1✔
513
514	func (rl *RedisCache) KeyCheckInTime(tid, id, idtype string, version int) string {	1✔
515	return fmt.Sprintf(	1✔
516	"%s:tenant:%s:version:%d:%s:%s:checkInTime",	1✔
517	rl.prefix, tid, version, idtype, id)	1✔
518	}	1✔
519
520	func (rl *RedisCache) KeyTenantVersion(tid string) string {	1✔
521	return fmt.Sprintf("%s:tenant:%s:version", rl.prefix, tid)	1✔
522	}	1✔
523
524	// isErrRedisNil checks for a very common non-error, "redis: nil",
525	// which just means the key was not found, and is normal
526	// it's routinely returned e.g. from GET, or pipelines containing it
527	func isErrRedisNil(e error) bool {	1✔
528	return errors.Is(e, redis.Nil)	1✔
529	}	1✔
530
531	// TODO: move to go-lib-micro/ratelimits
532	func LimitsEmpty(l *ratelimits.ApiLimits) bool {	×
533	return l.ApiQuota.MaxCalls == 0 &&	×
534	l.ApiQuota.IntervalSec == 0 &&	×
535	len(l.ApiBursts) == 0	×
536	}	×
537
538	func (rl *RedisCache) CacheCheckInTime(
539	ctx context.Context,
540	t *time.Time,
541	tid,
542	id string,
543	) error {	1✔
544	tj, err := json.Marshal(t)	1✔
545	if err != nil {	1✔
546	return err	×
547	}	×
548
549	version, err := rl.getTenantKeyVersion(ctx, tid)	1✔
550	if err != nil {	1✔
551	return err	×
552	}	×
553
554	res := rl.c.Set(	1✔
555	ctx,	1✔
556	rl.KeyCheckInTime(tid, id, IdTypeDevice, version),	1✔
557	tj,	1✔
558	CheckInTimeExpiration,	1✔
559	)	1✔
560		1✔
561	return res.Err()	1✔
562	}
563
564	func (rl *RedisCache) GetCheckInTime(
565	ctx context.Context,
566	tid,
567	id string,
568	) (*time.Time, error) {	1✔
569	version, err := rl.getTenantKeyVersion(ctx, tid)	1✔
570	if err != nil {	1✔
571	return nil, err	×
572	}	×
573
574	res := rl.c.Get(ctx, rl.KeyCheckInTime(tid, id, IdTypeDevice, version))	1✔
575		1✔
576	if res.Err() != nil {	2✔
577	if isErrRedisNil(res.Err()) {	2✔
578	return nil, nil	1✔
579	}	1✔
580	return nil, res.Err()	×
581	}
582
583	var checkInTime time.Time	1✔
584		1✔
585	err = json.Unmarshal([]byte(res.Val()), &checkInTime)	1✔
586	if err != nil {	1✔
587	return nil, err	×
588	}	×
589
590	return &checkInTime, nil	1✔
591	}
592
593	func (rl *RedisCache) GetCheckInTimes(
594	ctx context.Context,
595	tid string,
596	ids []string,
597	) ([]*time.Time, error) {	1✔
598	l := log.FromContext(ctx)	1✔
599		1✔
600	version, err := rl.getTenantKeyVersion(ctx, tid)	1✔
601	if err != nil {	1✔
602	return nil, err	×
603	}	×
604	checkInTimes := make([]*time.Time, len(ids))	1✔
605	if _, ok := rl.c.(*redis.ClusterClient); ok {	1✔
606	pipe := rl.c.Pipeline()	×
607	for _, id := range ids {	×
608	pipe.Get(ctx, rl.KeyCheckInTime(tid, id, IdTypeDevice, version))	×
609	}	×
610	results, err := pipe.Exec(ctx)	×
611	if err != nil && !errors.Is(err, redis.Nil) {	×
612	return nil, fmt.Errorf("failed to fetch check in times: %w", err)	×
613	}	×
614	for i, result := range results {	×
615	cmd, ok := result.(*redis.StringCmd)	×
616	if !ok {	×
617	continue // should never happen	×
618	}
619	b, err := cmd.Bytes()	×
620	if err != nil {	×
621	if errors.Is(err, redis.Nil) {	×
622	continue	×
623	} else {	×
624	l.Errorf("failed to get device: %s", err.Error())	×
625	}	×
626	} else {	×
627	checkInTime := new(time.Time)	×
628	err = json.Unmarshal(b, checkInTime)	×
629	if err != nil {	×
630	l.Errorf("failed to deserialize check in time: %s", err.Error())	×
631	} else {	×
632	checkInTimes[i] = checkInTime	×
633	}	×
634
635	}
636	}
637	} else {	1✔
638	keys := make([]string, len(ids))	1✔
639	for i, id := range ids {	2✔
640	keys[i] = rl.KeyCheckInTime(tid, id, IdTypeDevice, version)	1✔
641	}	1✔
642	res := rl.c.MGet(ctx, keys...)	1✔
643		1✔
644	for i, v := range res.Val() {	2✔
645	if v != nil {	2✔
646	b, ok := v.(string)	1✔
647	if !ok {	1✔
648	continue	×
649	}
650	var checkInTime time.Time	1✔
651	err := json.Unmarshal([]byte(b), &checkInTime)	1✔
652	if err != nil {	1✔
653	l.Errorf("failed to unmarshal check-in time: %s", err.Error())	×
654	continue	×
655	}
656	checkInTimes[i] = &checkInTime	1✔
657	}
658	}
659	}
660
661	return checkInTimes, nil	1✔
662	}
663
664	func (rl *RedisCache) SuspendTenant(
665	ctx context.Context,
666	tid string,
667	) error {	×
668	res := rl.c.Incr(ctx, rl.KeyTenantVersion(tid))	×
669	return res.Err()	×
670	}	×
671
672	func (rl *RedisCache) getTenantKeyVersion(ctx context.Context, tid string) (int, error) {	1✔
673	res := rl.c.Get(ctx, rl.KeyTenantVersion(tid))	1✔
674	if res.Err() != nil {	2✔
675	if isErrRedisNil(res.Err()) {	2✔
676	return 0, nil	1✔
677	}	1✔
678	return 0, res.Err()	×
679	}
680
681	var version int	×
682		×
683	err := json.Unmarshal([]byte(res.Val()), &version)	×
684	if err != nil {	×
685	return 0, err	×
686	}	×
687
688	return version, nil	×
689	}

mendersoftware / mender-server / 1550729677

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