Redis MOVED error causes shutdown during cluster leader change

2025/05/21 13:38:40.192 [E] MOVED 8127 10.128.0.107:11003

We are seeing this in productions before the instance shuts down and restarts. I think this is happening when redis or valkey changes owner in a cluster setup. We are using google memorystorage for example in cluster mode. Any ideas on how to mitigate this?

WHAT ARE YOU SAYING?

Wrong channel?

No. I think there is a bug in the redis implementation. That occurs when redis or valkey changes primary leader in the cluster, casdoor starts presenting that log and shuts down. I have no stack trace to show just that log line.
Log line: ``` MOVED 8127 10.128.0.107:11003``

The demo site also uses Redis 5, no such issue

Redis is used by Beego, even if you have question, you can resort to Beego

@hsluoyz Redis 5 is EOL so thats not really an option to run. If we need to remove redis, is there alternative to session management besides tmp files?

I see that Beego supports postgres for sessions but I dont see how to set that in Casdoor

https://github.com/casdoor/casdoor/blob/87506b84e3c9bdb02534a27f779096e25bfaded2/main.go#L74 Actually I think this is the issue. Can we expose this as a setting because beego has a redis_cluster provider as well as redis?

Where's the beego docs for "beego has a redis_cluster provider as well "?

can you create a github issue for all these?

we can track it there

https://github.com/beego/beego/blob/master/server/web/session/redis_cluster/redis_cluster.go#L24

Yes. I will write one up tonight.

your link is Beego v2, but Casdoor is Beego v1 (we don't want to upgrade to v2)

Why do we not want to upgrade?

It’s also in the develop-1x branch. https://github.com/beego/beego/blob/develop-1.x/session/redis_cluster/redis_cluster.go

need it to be in our upgrade path

means we 1.2.3 upgrade to 1.2.4, then have it

no lots of cherry-pick/back-port stuff

no explains for 2.0, find github clue by yourself if interested

What you are saying doesn’t make sense because it’s in the latest 1x release 🤷🏻

I pulled v1.12.12 which casdoor is using down and this file is present in there as well. So i am not following you when you say its "2.0" only.

redis_cluster.go:

// Copyright 2014 beego Author. All Rights Reserved.
//
// 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 redis for session provider
//
// depend on github.com/go-redis/redis
//
// go install github.com/go-redis/redis
//
// Usage:
// import(
//   _ "github.com/beego/beego/session/redis_cluster"
//   "github.com/beego/beego/session"
// )
//
//	func init() {
//		globalSessions, _ = session.NewManager("redis_cluster", ``{"cookieName":"gosessionid","gclifetime":3600,"ProviderConfig":"127.0.0.1:7070;127.0.0.1:7071"}``)
//		go globalSessions.GC()
//	}
//
// more docs: http://beego.me/docs/module/session.md
package redis_cluster

import (
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/beego/beego/session"
	rediss "github.com/go-redis/redis"
)

var redispder = &Provider{}

// MaxPoolSize redis_cluster max pool size
var MaxPoolSize = 1000

// SessionStore redis_cluster session store
type SessionStore struct {
	p           *rediss.ClusterClient
	sid         string
	lock        sync.RWMutex
	values      map[interface{}]interface{}
	maxlifetime int64
}

// Set value in redis_cluster session
func (rs *SessionStore) Set(key, value interface{}) error {
	rs.lock.Lock()
	defer rs.lock.Unlock()
	rs.values[key] = value
	return nil
}

// Get value in redis_cluster session
func (rs *SessionStore) Get(key interface{}) interface{} {
	rs.lock.RLock()
	defer rs.lock.RUnlock()
	if v, ok := rs.values[key]; ok {
		return v
	}
	return nil
}

// Delete value in redis_cluster session
func (rs *SessionStore) Delete(key interface{}) error {
	rs.lock.Lock()
	defer rs.lock.Unlock()
	delete(rs.values, key)
	return nil
}

// Flush clear all values in redis_cluster session
func (rs *SessionStore) Flush() error {
	rs.lock.Lock()
	defer rs.lock.Unlock()
	rs.values = make(map[interface{}]interface{})
	return nil
}

// SessionID get redis_cluster session id
func (rs *SessionStore) SessionID() string {
	return rs.sid
}

// SessionRelease save session values to redis_cluster
func (rs *SessionStore) SessionRelease(w http.ResponseWriter) {
	b, err := session.EncodeGob(rs.values)
	if err != nil {
		return
	}
	c := rs.p
	c.Set(rs.sid, string(b), time.Duration(rs.maxlifetime)*time.Second)
}

// Provider redis_cluster session provider
type Provider struct {
	maxlifetime int64
	savePath    string
	poolsize    int
	password    string
	dbNum       int
	poollist    *rediss.ClusterClient
}

// SessionInit init redis_cluster session
// savepath like redis server addr,pool size,password,dbnum
// e.g. 127.0.0.1:6379;127.0.0.1:6380,100,test,0
func (rp *Provider) SessionInit(maxlifetime int64, savePath string) error {
	rp.maxlifetime = maxlifetime
	configs := strings.Split(savePath, ",")
	if len(configs) > 0 {
		rp.savePath = configs[0]
	}
	if len(configs) > 1 {
		poolsize, err := strconv.Atoi(configs[1])
		if err != nil || poolsize < 0 {
			rp.poolsize = MaxPoolSize
		} else {
			rp.poolsize = poolsize
		}
	} else {
		rp.poolsize = MaxPoolSize
	}
	if len(configs) > 2 {
		rp.password = ***
	}
	if len(configs) > 3 {
		dbnum, err := strconv.Atoi(configs[3])
		if err != nil || dbnum < 0 {
			rp.dbNum = 0
		} else {
			rp.dbNum = dbnum
		}
	} else {
		rp.dbNum = 0
	}

	rp.poollist = rediss.NewClusterClient(&rediss.ClusterOptions{
		Addrs:    strings.Split(rp.savePath, ";"),
		Password: ***,
		PoolSize: rp.poolsize,
	})
	return rp.poollist.Ping().Err()
}

// SessionRead read redis_cluster session by sid
func (rp *Provider) SessionRead(sid string) (session.Store, error) {
	var kv map[interface{}]interface{}
	kvs, err := rp.poollist.Get(sid).Result()
	if err != nil && err != rediss.Nil {
		return nil, err
	}
	if len(kvs) == 0 {
		kv = make(map[interface{}]interface{})
	} else {
		if kv, err = session.DecodeGob([]byte(kvs)); err != nil {
			return nil, err
		}
	}

	rs := &SessionStore{p: rp.poollist, sid: sid, values: kv, maxlifetime: rp.maxlifetime}
	return rs, nil
}

// SessionExist check redis_cluster session exist by sid
func (rp *Provider) SessionExist(sid string) bool {
	c := rp.poollist
	if existed, err := c.Exists(sid).Result(); err != nil || existed == 0 {
		return false
	}
	return true
}

// SessionRegenerate generate new sid for redis_cluster session
func (rp *Provider) SessionRegenerate(oldsid, sid string) (session.Store, error) {
	c := rp.poollist

	if existed, err := c.Exists(oldsid).Result(); err != nil || existed == 0 {
		// oldsid doesn't exists, set the new sid directly
		// ignore error here, since if it return error
		// the existed value will be 0
		c.Set(sid, "", time.Duration(rp.maxlifetime)*time.Second)
	} else {
		c.Rename(oldsid, sid)
		c.Expire(sid, time.Duration(rp.maxlifetime)*time.Second)
	}
	return rp.SessionRead(sid)
}

// SessionDestroy delete redis session by id
func (rp *Provider) SessionDestroy(sid string) error {
	c := rp.poollist
	c.Del(sid)
	return nil
}

// SessionGC Impelment method, no used.
func (rp *Provider) SessionGC() {
}

// SessionAll return all activeSession
func (rp *Provider) SessionAll() int {
	return 0
}

func init() {
	session.Register("redis_cluster", redispder)
}

that's good!

So the Beego version used by Casdoor now already has it, right?

@oscillatingdecay can you make a PR in Casdoor repo to support that kind of redis cluster?

Yes, I am working on this over the next few days. Should we add the other options to use databases like mysql