package server import ( "errors" "fmt" "io" "strconv" "strings" "time" "github.com/tidwall/resp" "github.com/tidwall/tile38/internal/log" ) var errNoLongerFollowing = errors.New("no longer following") const checksumsz = 512 * 1024 func (s *Server) cmdFollow(msg *Message) (res resp.Value, err error) { start := time.Now() vs := msg.Args[1:] var ok bool var host, sport string if vs, host, ok = tokenval(vs); !ok || host == "" { return NOMessage, errInvalidNumberOfArguments } if vs, sport, ok = tokenval(vs); !ok || sport == "" { return NOMessage, errInvalidNumberOfArguments } if len(vs) != 0 { return NOMessage, errInvalidNumberOfArguments } host = strings.ToLower(host) sport = strings.ToLower(sport) var update bool if host == "no" && sport == "one" { update = s.config.followHost() != "" || s.config.followPort() != 0 s.config.setFollowHost("") s.config.setFollowPort(0) } else { n, err := strconv.ParseUint(sport, 10, 64) if err != nil { return NOMessage, errInvalidArgument(sport) } port := int(n) update = s.config.followHost() != host || s.config.followPort() != port auth := s.config.leaderAuth() if update { s.mu.Unlock() conn, err := DialTimeout(fmt.Sprintf("%s:%d", host, port), time.Second*2) if err != nil { s.mu.Lock() return NOMessage, fmt.Errorf("cannot follow: %v", err) } defer conn.Close() if auth != "" { if err := s.followDoLeaderAuth(conn, auth); err != nil { return NOMessage, fmt.Errorf("cannot follow: %v", err) } } m, err := doServer(conn) if err != nil { s.mu.Lock() return NOMessage, fmt.Errorf("cannot follow: %v", err) } if m["id"] == "" { s.mu.Lock() return NOMessage, fmt.Errorf("cannot follow: invalid id") } if m["id"] == s.config.serverID() { s.mu.Lock() return NOMessage, fmt.Errorf("cannot follow self") } if m["following"] != "" { s.mu.Lock() return NOMessage, fmt.Errorf("cannot follow a follower") } s.mu.Lock() } s.config.setFollowHost(host) s.config.setFollowPort(port) } s.config.write(false) if update { s.followc.Add(1) if s.config.followHost() != "" { log.Infof("following new host '%s' '%s'.", host, sport) go s.follow(s.config.followHost(), s.config.followPort(), int(s.followc.Load())) } else { log.Infof("following no one") } } return OKMessage(msg, start), nil } // cmdReplConf is a command handler that sets replication configuration info func (s *Server) cmdReplConf(msg *Message, client *Client) (res resp.Value, err error) { start := time.Now() vs := msg.Args[1:] var ok bool var cmd, val string // Parse the message if vs, cmd, ok = tokenval(vs); !ok || cmd == "" { return NOMessage, errInvalidNumberOfArguments } if _, val, ok = tokenval(vs); !ok || val == "" { return NOMessage, errInvalidNumberOfArguments } // Switch on the command received switch cmd { case "listening-port": // Parse the port as an integer port, err := strconv.Atoi(val) if err != nil { return NOMessage, errInvalidArgument(val) } // Apply the replication port to the client and return s.connsmu.RLock() defer s.connsmu.RUnlock() for _, c := range s.conns { if c.remoteAddr == client.remoteAddr { c.mu.Lock() c.replPort = port c.mu.Unlock() return OKMessage(msg, start), nil } } case "ip-address": // Apply the replication ip to the client and return s.connsmu.RLock() defer s.connsmu.RUnlock() for _, c := range s.conns { if c.remoteAddr == client.remoteAddr { c.mu.Lock() c.replAddr = val c.mu.Unlock() return OKMessage(msg, start), nil } } } return NOMessage, fmt.Errorf("cannot find follower") } func doServer(conn *RESPConn) (map[string]string, error) { v, err := conn.Do("server") if err != nil { return nil, err } if v.Error() != nil { return nil, v.Error() } arr := v.Array() m := make(map[string]string) for i := 0; i < len(arr)/2; i++ { m[arr[i*2+0].String()] = arr[i*2+1].String() } return m, err } func (s *Server) followHandleCommand(args []string, followc int, w io.Writer) (int, error) { s.mu.Lock() defer s.mu.Unlock() if int(s.followc.Load()) != followc { return s.aofsz, errNoLongerFollowing } msg := &Message{Args: args} _, d, err := s.command(msg, nil) if err != nil { if commandErrIsFatal(err) { return s.aofsz, err } } switch msg.Command() { case "publish": // Avoid writing these commands to the AOF default: if err := s.writeAOF(args, &d); err != nil { return s.aofsz, err } } if len(s.aofbuf) > 10240 { s.flushAOF(false) } return s.aofsz, nil } func (s *Server) followDoLeaderAuth(conn *RESPConn, auth string) error { v, err := conn.Do("auth", auth) if err != nil { return err } if v.Error() != nil { return v.Error() } if v.String() != "OK" { return errors.New("cannot follow: auth no ok") } return nil } func (s *Server) followStep(host string, port int, followc int) error { if int(s.followc.Load()) != followc { return errNoLongerFollowing } s.mu.Lock() s.faofsz = 0 s.fcup = false auth := s.config.leaderAuth() s.mu.Unlock() addr := fmt.Sprintf("%s:%d", host, port) // check if we are following self conn, err := DialTimeout(addr, time.Second*2) if err != nil { return fmt.Errorf("cannot follow: %v", err) } defer conn.Close() if auth != "" { if err := s.followDoLeaderAuth(conn, auth); err != nil { return fmt.Errorf("cannot follow: %v", err) } } m, err := doServer(conn) if err != nil { return fmt.Errorf("cannot follow: %v", err) } if m["id"] == "" { return fmt.Errorf("cannot follow: invalid id") } if m["id"] == s.config.serverID() { return fmt.Errorf("cannot follow self") } if m["following"] != "" { return fmt.Errorf("cannot follow a follower") } // verify checksum pos, err := s.followCheckSome(addr, followc, auth) if err != nil { return err } // Send the replication port to the leader p := s.config.announcePort() if p == 0 { p = s.port } v, err := conn.Do("replconf", "listening-port", p) if err != nil { return err } if v.Error() != nil { return v.Error() } if v.String() != "OK" { return errors.New("invalid response to replconf request") } // Send the replication ip to the leader ip := s.config.announceIP() if ip != "" { v, err := conn.Do("replconf", "ip-address", ip) if err != nil { return err } if v.Error() != nil { return v.Error() } if v.String() != "OK" { return errors.New("invalid response to replconf request") } } if s.opts.ShowDebugMessages { log.Debug("follow:", addr, ":replconf") } v, err = conn.Do("aof", pos) if err != nil { return err } if v.Error() != nil { return v.Error() } if v.String() != "OK" { return errors.New("invalid response to aof live request") } if s.opts.ShowDebugMessages { log.Debug("follow:", addr, ":read aof") } aofSize, err := strconv.ParseInt(m["aof_size"], 10, 64) if err != nil { return err } s.mu.Lock() s.faofsz = int(aofSize) s.mu.Unlock() caughtUp := pos >= aofSize if caughtUp { s.mu.Lock() s.fcup = true s.fcuponce = true s.mu.Unlock() log.Info("caught up") } nullw := io.Discard for { v, telnet, _, err := conn.rd.ReadMultiBulk() if err != nil { return err } vals := v.Array() if telnet || v.Type() != resp.Array { return errors.New("invalid multibulk") } svals := make([]string, len(vals)) for i := 0; i < len(vals); i++ { svals[i] = vals[i].String() } aofsz, err := s.followHandleCommand(svals, followc, nullw) if err != nil { return err } s.mu.Lock() s.faofsz = aofsz s.mu.Unlock() if !caughtUp { if aofsz >= int(aofSize) { caughtUp = true s.mu.Lock() s.flushAOF(false) s.fcup = true s.fcuponce = true s.mu.Unlock() log.Info("caught up") } } } } func (s *Server) follow(host string, port int, followc int) { for { err := s.followStep(host, port, followc) if err == errNoLongerFollowing { return } if err != nil && err != io.EOF { log.Error("follow: " + err.Error()) } time.Sleep(time.Second) } }