Skip to content

Commit b60563b

Browse files
rawe0rawe0
authored
Redis sentinel mode support (#68)
1 parent 95a5b49 commit b60563b

5 files changed

Lines changed: 380 additions & 225 deletions

File tree

‎group/cache.go‎

Lines changed: 39 additions & 108 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// Copyright 2018-2023 CERN
1+
// Copyright 2018-2026 CERN
22
//
33
// Licensed under the Apache License, Version 2.0 (the "License");
44
// you may not use this file except in compliance with the License.
@@ -19,15 +19,15 @@
1919
package rest
2020

2121
import (
22+
"context"
2223
"encoding/json"
23-
"errors"
2424
"fmt"
2525
"strconv"
2626
"strings"
27-
"time"
2827

2928
grouppb "github.com/cs3org/go-cs3apis/cs3/identity/group/v1beta1"
3029
userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
30+
"github.com/cs3org/reva/v3/pkg/appctx"
3131
"github.com/gomodule/redigo/redis"
3232
)
3333

@@ -40,116 +40,47 @@ const (
4040
groupInternalIDPrefix = "internal:"
4141
)
4242

43-
func initRedisPool(address, username, password string) *redis.Pool {
44-
return &redis.Pool{
45-
46-
MaxIdle: 50,
47-
MaxActive: 1000,
48-
IdleTimeout: 240 * time.Second,
49-
50-
Dial: func() (redis.Conn, error) {
51-
var c redis.Conn
52-
var err error
53-
switch {
54-
case username != "":
55-
c, err = redis.Dial("tcp", address,
56-
redis.DialUsername(username),
57-
redis.DialPassword(password),
58-
)
59-
case password != "":
60-
c, err = redis.Dial("tcp", address,
61-
redis.DialPassword(password),
62-
)
63-
default:
64-
c, err = redis.Dial("tcp", address)
65-
}
66-
67-
if err != nil {
68-
return nil, err
69-
}
70-
return c, err
71-
},
72-
73-
TestOnBorrow: func(c redis.Conn, t time.Time) error {
74-
_, err := c.Do("PING")
75-
return err
76-
},
77-
}
78-
}
43+
func (m *manager) findCachedGroups(ctx context.Context, query string) ([]*grouppb.Group, error) {
44+
query = fmt.Sprintf("%s*%s*", groupPrefix, strings.ReplaceAll(strings.ToLower(query), " ", "_"))
45+
log := appctx.GetLogger(ctx)
7946

80-
func (m *manager) setVal(key, val string, expiration int) error {
81-
conn := m.redisPool.Get()
82-
defer conn.Close()
83-
if conn != nil {
84-
args := []interface{}{key, val}
85-
if expiration != -1 {
86-
args = append(args, "EX", expiration)
87-
}
88-
if _, err := conn.Do("SET", args...); err != nil {
89-
return err
90-
}
91-
return nil
92-
}
93-
return errors.New("rest: unable to get connection from redis pool")
94-
}
95-
96-
func (m *manager) getVal(key string) (string, error) {
97-
conn := m.redisPool.Get()
98-
defer conn.Close()
99-
if conn != nil {
100-
val, err := redis.String(conn.Do("GET", key))
101-
if err != nil {
102-
return "", err
103-
}
104-
return val, nil
105-
}
106-
return "", errors.New("rest: unable to get connection from redis pool")
107-
}
108-
109-
func (m *manager) findCachedGroups(query string) ([]*grouppb.Group, error) {
110-
conn := m.redisPool.Get()
111-
defer conn.Close()
112-
if conn != nil {
113-
query = fmt.Sprintf("%s*%s*", groupPrefix, strings.ReplaceAll(strings.ToLower(query), " ", "_"))
47+
raw, err := m.redisPools.DoWithReadFallback(ctx, func(conn redis.Conn) (interface{}, error) {
11448
keys, err := redis.Strings(conn.Do("KEYS", query))
11549
if err != nil {
11650
return nil, err
11751
}
118-
var args []interface{}
119-
for _, k := range keys {
120-
args = append(args, k)
121-
}
122-
123-
if len(args) == 0 {
124-
return []*grouppb.Group{}, nil
125-
}
126-
127-
// Fetch the groups for all these keys
128-
groupStrings, err := redis.Strings(conn.Do("MGET", args...))
129-
if err != nil {
130-
return nil, err
52+
if len(keys) == 0 {
53+
return []string{}, nil
13154
}
132-
groupMap := make(map[string]*grouppb.Group)
133-
for _, group := range groupStrings {
134-
g := grouppb.Group{}
135-
if err = json.Unmarshal([]byte(group), &g); err == nil {
136-
groupMap[g.Id.OpaqueId] = &g
137-
}
55+
args := make([]interface{}, len(keys))
56+
for i, k := range keys {
57+
args[i] = k
13858
}
59+
return redis.Strings(conn.Do("MGET", args...))
60+
})
61+
if err != nil {
62+
return nil, err
63+
}
13964

140-
var groups []*grouppb.Group
141-
for _, g := range groupMap {
142-
groups = append(groups, g)
65+
groupStrings := raw.([]string)
66+
groupMap := make(map[string]*grouppb.Group)
67+
for _, group := range groupStrings {
68+
g := grouppb.Group{}
69+
if err = json.Unmarshal([]byte(group), &g); err == nil {
70+
groupMap[g.Id.OpaqueId] = &g
14371
}
144-
145-
return groups, nil
14672
}
14773

148-
return nil, errors.New("rest: unable to get connection from redis pool")
74+
groups := make([]*grouppb.Group, 0, len(groupMap))
75+
for _, g := range groupMap {
76+
groups = append(groups, g)
77+
}
78+
log.Debug().Any("query", query).Int("results", len(groups)).Msg("rest: successfully found cached groups")
79+
return groups, nil
14980
}
15081

151-
func (m *manager) fetchCachedGroupDetails(gid *grouppb.GroupId) (*grouppb.Group, error) {
152-
group, err := m.getVal(groupPrefix + idPrefix + gid.OpaqueId)
82+
func (m *manager) fetchCachedGroupDetails(ctx context.Context, gid *grouppb.GroupId) (*grouppb.Group, error) {
83+
group, err := m.redisPools.GetVal(ctx, groupPrefix+idPrefix+gid.OpaqueId)
15384
if err != nil {
15485
return nil, err
15586
}
@@ -166,25 +97,25 @@ func (m *manager) cacheGroupDetails(g *grouppb.Group) error {
16697
if err != nil {
16798
return err
16899
}
169-
if err = m.setVal(groupPrefix+idPrefix+strings.ToLower(g.Id.OpaqueId), string(encodedGroup), 5*m.conf.GroupFetchInterval); err != nil {
100+
if err = m.redisPools.SetVal(groupPrefix+idPrefix+strings.ToLower(g.Id.OpaqueId), string(encodedGroup), 5*m.conf.GroupFetchInterval); err != nil {
170101
return err
171102
}
172103

173104
if g.GidNumber != 0 {
174-
if err = m.setVal(groupPrefix+gidPrefix+strconv.FormatInt(g.GidNumber, 10), g.Id.OpaqueId, 5*m.conf.GroupFetchInterval); err != nil {
105+
if err = m.redisPools.SetVal(groupPrefix+gidPrefix+strconv.FormatInt(g.GidNumber, 10), g.Id.OpaqueId, 5*m.conf.GroupFetchInterval); err != nil {
175106
return err
176107
}
177108
}
178109
if g.DisplayName != "" {
179-
if err = m.setVal(groupPrefix+namePrefix+g.Id.OpaqueId+"_"+strings.ToLower(g.DisplayName), g.Id.OpaqueId, 5*m.conf.GroupFetchInterval); err != nil {
110+
if err = m.redisPools.SetVal(groupPrefix+namePrefix+g.Id.OpaqueId+"_"+strings.ToLower(g.DisplayName), g.Id.OpaqueId, 5*m.conf.GroupFetchInterval); err != nil {
180111
return err
181112
}
182113
}
183114
return nil
184115
}
185116

186-
func (m *manager) fetchCachedGroupByParam(field, claim string) (*grouppb.Group, error) {
187-
group, err := m.getVal(groupPrefix + field + ":" + strings.ToLower(claim))
117+
func (m *manager) fetchCachedGroupByParam(ctx context.Context, field, claim string) (*grouppb.Group, error) {
118+
group, err := m.redisPools.GetVal(ctx, groupPrefix+field+":"+strings.ToLower(claim))
188119
if err != nil {
189120
return nil, err
190121
}
@@ -196,8 +127,8 @@ func (m *manager) fetchCachedGroupByParam(field, claim string) (*grouppb.Group,
196127
return &g, nil
197128
}
198129

199-
func (m *manager) fetchCachedGroupMembers(gid *grouppb.GroupId) ([]*userpb.UserId, error) {
200-
members, err := m.getVal(groupPrefix + groupMembersPrefix + strings.ToLower(gid.OpaqueId))
130+
func (m *manager) fetchCachedGroupMembers(ctx context.Context, gid *grouppb.GroupId) ([]*userpb.UserId, error) {
131+
members, err := m.redisPools.GetVal(ctx, groupPrefix+groupMembersPrefix+strings.ToLower(gid.OpaqueId))
201132
if err != nil {
202133
return nil, err
203134
}
@@ -213,5 +144,5 @@ func (m *manager) cacheGroupMembers(gid *grouppb.GroupId, members []*userpb.User
213144
if err != nil {
214145
return err
215146
}
216-
return m.setVal(groupPrefix+groupMembersPrefix+strings.ToLower(gid.OpaqueId), string(u), m.conf.GroupMembersCacheExpiration*60)
147+
return m.redisPools.SetVal(groupPrefix+groupMembersPrefix+strings.ToLower(gid.OpaqueId), string(u), m.conf.GroupMembersCacheExpiration*60)
217148
}

‎group/rest.go‎

Lines changed: 23 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// Copyright 2018-2023 CERN
1+
// Copyright 2018-2026 CERN
22
//
33
// Licensed under the Apache License, Version 2.0 (the "License");
44
// you may not use this file except in compliance with the License.
@@ -27,6 +27,7 @@ import (
2727
"syscall"
2828
"time"
2929

30+
redispools "github.com/cernbox/reva-plugins/redispools"
3031
user "github.com/cernbox/reva-plugins/user"
3132
grouppb "github.com/cs3org/go-cs3apis/cs3/identity/group/v1beta1"
3233
userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
@@ -36,7 +37,6 @@ import (
3637
"github.com/cs3org/reva/v3/pkg/group"
3738
"github.com/cs3org/reva/v3/pkg/utils/cfg"
3839
"github.com/cs3org/reva/v3/pkg/utils/list"
39-
"github.com/gomodule/redigo/redis"
4040
"github.com/rs/zerolog/log"
4141
)
4242

@@ -46,7 +46,7 @@ func init() {
4646

4747
type manager struct {
4848
conf *config
49-
redisPool *redis.Pool
49+
redisPools *redispools.RedisPools
5050
apiTokenManager *utils.APITokenManager
5151
}
5252

@@ -60,10 +60,16 @@ func (manager) RevaPlugin() reva.PluginInfo {
6060
type config struct {
6161
// The address at which the redis server is running
6262
RedisAddress string `mapstructure:"redis_address" docs:"localhost:6379"`
63+
// The address at which the redis sentinel is running (only used when redis_sentinel_mode is enabled)
64+
RedisSentinelAddress string `mapstructure:"redis_sentinel_address" docs:""`
6365
// The username for connecting to the redis server
6466
RedisUsername string `mapstructure:"redis_username" docs:""`
6567
// The password for connecting to the redis server
6668
RedisPassword string `mapstructure:"redis_password" docs:""`
69+
// The name of the master node in case Redis Sentinel mode is enabled
70+
RedisMasterName string `mapstructure:"redis_master_name" docs:""`
71+
// Whether to use Redis Sentinel mode
72+
RedisSentinelMode bool `mapstructure:"redis_sentinel_mode" docs:""`
6773
// The time in minutes for which the members of a group would be cached
6874
GroupMembersCacheExpiration int `mapstructure:"group_members_cache_expiration" docs:"5"`
6975
// The OIDC Provider
@@ -90,6 +96,9 @@ func (c *config) ApplyDefaults() {
9096
if c.RedisAddress == "" {
9197
c.RedisAddress = ":6379"
9298
}
99+
if c.RedisSentinelAddress == "" {
100+
c.RedisSentinelAddress = c.RedisAddress
101+
}
93102
if c.APIBaseURL == "" {
94103
c.APIBaseURL = "https://authorization-service-api-dev.web.cern.ch"
95104
}
@@ -113,16 +122,21 @@ func New(ctx context.Context, m map[string]interface{}) (group.Manager, error) {
113122
if err := cfg.Decode(m, &c); err != nil {
114123
return nil, err
115124
}
125+
c.ApplyDefaults()
116126

117-
redisPool := initRedisPool(c.RedisAddress, c.RedisUsername, c.RedisPassword)
127+
pools, err := redispools.NewRedisPoolsWithSentinelAddress(ctx, c.RedisAddress, c.RedisSentinelAddress, c.RedisUsername, c.RedisPassword, c.RedisSentinelMode, c.RedisMasterName)
128+
if err != nil {
129+
appctx.GetLogger(ctx).Error().Err(err).Msg("rest: failed to initialize redis pools")
130+
pools = &redispools.RedisPools{}
131+
}
118132
apiTokenManager, err := utils.InitAPITokenManager(m)
119133
if err != nil {
120134
return nil, err
121135
}
122136

123137
mgr := &manager{
124138
conf: &c,
125-
redisPool: redisPool,
139+
redisPools: pools,
126140
apiTokenManager: apiTokenManager,
127141
}
128142
go mgr.fetchAllGroups(context.Background())
@@ -212,7 +226,7 @@ func (m *manager) parseAndCacheGroup(ctx context.Context, g *Group) (*grouppb.Gr
212226
}
213227

214228
func (m *manager) GetGroup(ctx context.Context, gid *grouppb.GroupId, skipFetchingMembers bool) (*grouppb.Group, error) {
215-
g, err := m.fetchCachedGroupDetails(gid)
229+
g, err := m.fetchCachedGroupDetails(ctx, gid)
216230
if err != nil {
217231
return nil, err
218232
}
@@ -233,7 +247,7 @@ func (m *manager) GetGroupByClaim(ctx context.Context, claim, value string, skip
233247
return m.GetGroup(ctx, &grouppb.GroupId{OpaqueId: value}, skipFetchingMembers)
234248
}
235249

236-
g, err := m.fetchCachedGroupByParam(claim, value)
250+
g, err := m.fetchCachedGroupByParam(ctx, claim, value)
237251
if err != nil {
238252
return nil, err
239253
}
@@ -264,11 +278,11 @@ func (m *manager) FindGroups(ctx context.Context, query string, skipFetchingMemb
264278
}
265279
}
266280

267-
return m.findCachedGroups(query)
281+
return m.findCachedGroups(ctx, query)
268282
}
269283

270284
func (m *manager) GetMembers(ctx context.Context, gid *grouppb.GroupId) ([]*userpb.UserId, error) {
271-
users, err := m.fetchCachedGroupMembers(gid)
285+
users, err := m.fetchCachedGroupMembers(ctx, gid)
272286
if err == nil {
273287
return users, nil
274288
}

0 commit comments

Comments
 (0)