mirror of
https://github.com/etcd-io/etcd.git
synced 2024-09-27 06:25:44 +00:00
fix redirect after seprate client and server pot
This commit is contained in:
parent
430f5d50e3
commit
c2f436a58b
12
command.go
12
command.go
@ -16,7 +16,7 @@ import (
|
||||
// A command represents an action to be taken on the replicated state machine.
|
||||
type Command interface {
|
||||
CommandName() string
|
||||
Apply(server *raft.Server) (interface {}, error)
|
||||
Apply(server *raft.Server) (interface{}, error)
|
||||
}
|
||||
|
||||
// Set command
|
||||
@ -32,7 +32,7 @@ func (c *SetCommand) CommandName() string {
|
||||
}
|
||||
|
||||
// Set the value of key to value
|
||||
func (c *SetCommand) Apply(server *raft.Server) (interface {}, error) {
|
||||
func (c *SetCommand) Apply(server *raft.Server) (interface{}, error) {
|
||||
return store.Set(c.Key, c.Value, c.ExpireTime)
|
||||
}
|
||||
|
||||
@ -52,7 +52,7 @@ func (c *GetCommand) CommandName() string {
|
||||
}
|
||||
|
||||
// Set the value of key to value
|
||||
func (c *GetCommand) Apply(server *raft.Server) (interface {}, error) {
|
||||
func (c *GetCommand) Apply(server *raft.Server) (interface{}, error) {
|
||||
res := store.Get(c.Key)
|
||||
return json.Marshal(res)
|
||||
}
|
||||
@ -72,7 +72,7 @@ func (c *DeleteCommand) CommandName() string {
|
||||
}
|
||||
|
||||
// Delete the key
|
||||
func (c *DeleteCommand) Apply(server *raft.Server) (interface {}, error) {
|
||||
func (c *DeleteCommand) Apply(server *raft.Server) (interface{}, error) {
|
||||
return store.Delete(c.Key)
|
||||
}
|
||||
|
||||
@ -86,7 +86,7 @@ func (c *WatchCommand) CommandName() string {
|
||||
return "watch"
|
||||
}
|
||||
|
||||
func (c *WatchCommand) Apply(server *raft.Server) (interface {}, error) {
|
||||
func (c *WatchCommand) Apply(server *raft.Server) (interface{}, error) {
|
||||
ch := make(chan store.Response)
|
||||
|
||||
// add to the watchers list
|
||||
@ -107,7 +107,7 @@ func (c *JoinCommand) CommandName() string {
|
||||
return "join"
|
||||
}
|
||||
|
||||
func (c *JoinCommand) Apply(server *raft.Server) (interface {}, error) {
|
||||
func (c *JoinCommand) Apply(server *raft.Server) (interface{}, error) {
|
||||
err := server.AddPeer(c.Name)
|
||||
// no result will be returned
|
||||
return nil, err
|
||||
|
20
etcd.go
20
etcd.go
@ -68,7 +68,7 @@ func init() {
|
||||
|
||||
flag.StringVar(&dirPath, "d", "./", "the directory to store log and snapshot")
|
||||
|
||||
flag.BoolVar(&ignore, "i", false , "ignore the old configuration, create a new node")
|
||||
flag.BoolVar(&ignore, "i", false, "ignore the old configuration, create a new node")
|
||||
}
|
||||
|
||||
// CONSTANTS
|
||||
@ -95,11 +95,10 @@ const (
|
||||
//------------------------------------------------------------------------------
|
||||
|
||||
type Info struct {
|
||||
Address string `json:"address"`
|
||||
Address string `json:"address"`
|
||||
ServerPort int `json:"serverPort"`
|
||||
ClientPort int `json:"clientPort"`
|
||||
WebPort int `json:"webPort"`
|
||||
|
||||
ClientPort int `json:"clientPort"`
|
||||
WebPort int `json:"webPort"`
|
||||
}
|
||||
|
||||
//------------------------------------------------------------------------------
|
||||
@ -109,6 +108,7 @@ type Info struct {
|
||||
//------------------------------------------------------------------------------
|
||||
|
||||
var server *raft.Server
|
||||
var serverTransHandler transHandler
|
||||
var logger *log.Logger
|
||||
|
||||
var storeMsg chan string
|
||||
@ -152,13 +152,13 @@ func main() {
|
||||
panic("ERROR type")
|
||||
}
|
||||
|
||||
t := createTranHandler(st)
|
||||
serverTransHandler = createTranHandler(st)
|
||||
|
||||
// Setup new raft server.
|
||||
s := store.GetStore()
|
||||
|
||||
// create raft server
|
||||
server, err = raft.NewServer(name, dirPath, t, s, nil)
|
||||
server, err = raft.NewServer(name, dirPath, serverTransHandler, s, nil)
|
||||
|
||||
if err != nil {
|
||||
fatal("%v", err)
|
||||
@ -262,7 +262,7 @@ func startServTransport(port int, st int) {
|
||||
http.HandleFunc("/log", GetLogHttpHandler)
|
||||
http.HandleFunc("/log/append", AppendEntriesHttpHandler)
|
||||
http.HandleFunc("/snapshot", SnapshotHttpHandler)
|
||||
|
||||
http.HandleFunc("/client", clientHttpHandler)
|
||||
|
||||
switch st {
|
||||
|
||||
@ -351,14 +351,13 @@ func startClientTransport(port int, st int) {
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
//--------------------------------------
|
||||
// Config
|
||||
//--------------------------------------
|
||||
|
||||
func securityType(source int) int {
|
||||
|
||||
var keyFile, certFile, CAFile string
|
||||
var keyFile, certFile, CAFile string
|
||||
|
||||
switch source {
|
||||
case SERVER:
|
||||
@ -480,4 +479,3 @@ func Join(s *raft.Server, serverName string) error {
|
||||
}
|
||||
return fmt.Errorf("Unable to join: %v", err)
|
||||
}
|
||||
|
||||
|
38
handlers.go
38
handlers.go
@ -13,7 +13,7 @@ import (
|
||||
)
|
||||
|
||||
//--------------------------------------
|
||||
// HTTP Handlers
|
||||
// Internal HTTP Handlers via server port
|
||||
//--------------------------------------
|
||||
|
||||
// Get all the current logs
|
||||
@ -72,6 +72,13 @@ func SnapshotHttpHandler(w http.ResponseWriter, req *http.Request) {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
func clientHttpHandler(w http.ResponseWriter, req *http.Request) {
|
||||
debug("[recv] Get http://%v/client/ ", server.Name())
|
||||
w.WriteHeader(http.StatusOK)
|
||||
client := address + ":" + strconv.Itoa(clientPort)
|
||||
w.Write([]byte(client))
|
||||
}
|
||||
|
||||
func JoinHttpHandler(w http.ResponseWriter, req *http.Request) {
|
||||
|
||||
command := &JoinCommand{}
|
||||
@ -85,6 +92,9 @@ func JoinHttpHandler(w http.ResponseWriter, req *http.Request) {
|
||||
}
|
||||
}
|
||||
|
||||
//--------------------------------------
|
||||
// external HTTP Handlers via client port
|
||||
//--------------------------------------
|
||||
func SetHttpHandler(w http.ResponseWriter, req *http.Request) {
|
||||
key := req.URL.Path[len("/set/"):]
|
||||
|
||||
@ -141,12 +151,12 @@ func excute(c Command, w *http.ResponseWriter, req *http.Request) {
|
||||
(*w).WriteHeader(http.StatusOK)
|
||||
|
||||
if body == nil {
|
||||
return
|
||||
return
|
||||
}
|
||||
|
||||
body, ok := body.([]byte)
|
||||
if !ok {
|
||||
panic ("wrong type")
|
||||
panic("wrong type")
|
||||
}
|
||||
|
||||
(*w).Write(body)
|
||||
@ -154,19 +164,19 @@ func excute(c Command, w *http.ResponseWriter, req *http.Request) {
|
||||
}
|
||||
} else {
|
||||
// tell the client where is the leader
|
||||
debug("Redirect to the leader %s", server.Leader())
|
||||
debug("Redirect to the leader %s", server.Leader())
|
||||
|
||||
path := req.URL.Path
|
||||
path := req.URL.Path
|
||||
|
||||
var scheme string
|
||||
var scheme string
|
||||
|
||||
if scheme = req.URL.Scheme; scheme == "" {
|
||||
scheme = "http://"
|
||||
}
|
||||
if scheme = req.URL.Scheme; scheme == "" {
|
||||
scheme = "http://"
|
||||
}
|
||||
|
||||
url := scheme + server.Leader() + path
|
||||
url := scheme + leaderClient() + path
|
||||
|
||||
debug("redirect to ", url)
|
||||
debug("redirect to ", url)
|
||||
http.Redirect(*w, req, url, http.StatusTemporaryRedirect)
|
||||
return
|
||||
}
|
||||
@ -198,7 +208,7 @@ func GetHttpHandler(w http.ResponseWriter, req *http.Request) {
|
||||
|
||||
body, ok := body.([]byte)
|
||||
if !ok {
|
||||
panic ("wrong type")
|
||||
panic("wrong type")
|
||||
}
|
||||
|
||||
w.Write(body)
|
||||
@ -224,9 +234,9 @@ func WatchHttpHandler(w http.ResponseWriter, req *http.Request) {
|
||||
|
||||
body, ok := body.([]byte)
|
||||
if !ok {
|
||||
panic ("wrong type")
|
||||
panic("wrong type")
|
||||
}
|
||||
|
||||
|
||||
w.Write(body)
|
||||
return
|
||||
}
|
||||
|
@ -109,7 +109,7 @@ func Set(key string, value string, expireTime time.Time) ([]byte, error) {
|
||||
go expire(key, node.update, expireTime)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// update the information of the node
|
||||
s.Nodes[key] = Node{value, expireTime, node.update}
|
||||
|
||||
|
@ -15,7 +15,7 @@ type WatcherHub struct {
|
||||
}
|
||||
|
||||
type Watcher struct {
|
||||
c chan Response
|
||||
c chan Response
|
||||
wType int
|
||||
}
|
||||
|
||||
@ -46,7 +46,7 @@ func AddWatcher(prefix string, c chan Response, wType int) error {
|
||||
_, ok := w.watchers[prefix]
|
||||
|
||||
if !ok {
|
||||
|
||||
|
||||
w.watchers[prefix] = make([]Watcher, 0)
|
||||
|
||||
watcher := Watcher{c, wType}
|
||||
@ -80,10 +80,10 @@ func notify(resp Response) error {
|
||||
newWatchers := make([]Watcher, 0)
|
||||
// notify all the watchers
|
||||
for _, watcher := range watchers {
|
||||
watcher.c <- resp
|
||||
if watcher.wType == LONG {
|
||||
newWatchers = append(newWatchers, watcher)
|
||||
}
|
||||
watcher.c <- resp
|
||||
if watcher.wType == LONG {
|
||||
newWatchers = append(newWatchers, watcher)
|
||||
}
|
||||
}
|
||||
|
||||
if len(newWatchers) == 0 {
|
||||
|
21
util.go
21
util.go
@ -1,13 +1,15 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"io"
|
||||
"fmt"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/xiangli-cmu/raft-etcd/web"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"os"
|
||||
)
|
||||
|
||||
//--------------------------------------
|
||||
// Web Helper
|
||||
//--------------------------------------
|
||||
@ -63,6 +65,17 @@ func Get(t *transHandler, path string) (*http.Response, error) {
|
||||
}
|
||||
}
|
||||
|
||||
func leaderClient() string {
|
||||
resp, _ := Get(&serverTransHandler, server.Leader()+"/client")
|
||||
if resp != nil {
|
||||
defer resp.Body.Close()
|
||||
body, _ := ioutil.ReadAll(resp.Body)
|
||||
|
||||
return string(body)
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
//--------------------------------------
|
||||
// Log
|
||||
//--------------------------------------
|
||||
@ -84,4 +97,4 @@ func warn(msg string, v ...interface{}) {
|
||||
func fatal(msg string, v ...interface{}) {
|
||||
logger.Printf("FATAL "+msg+"\n", v...)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
Loading…
x
Reference in New Issue
Block a user