package main import ( "encoding/json" "fmt" "io" "net" "net/http" "net/url" "os" "os/signal" "runtime/pprof" "strconv" "time" etcdErr "github.com/coreos/etcd/error" "github.com/coreos/etcd/store" "github.com/coreos/etcd/web" "github.com/coreos/go-log/log" "github.com/coreos/go-raft" ) //-------------------------------------- // etcd http Helper //-------------------------------------- // Convert string duration to time format func durationToExpireTime(strDuration string) (time.Time, error) { if strDuration != "" { duration, err := strconv.Atoi(strDuration) if err != nil { return store.Permanent, err } return time.Now().Add(time.Second * (time.Duration)(duration)), nil } else { return store.Permanent, nil } } //-------------------------------------- // Web Helper //-------------------------------------- var storeMsg chan string // Help to send msg from store to webHub func webHelper() { storeMsg = make(chan string) // etcdStore.SetMessager(storeMsg) for { // transfer the new msg to webHub web.Hub().Send(<-storeMsg) } } // startWebInterface starts web interface if webURL is not empty func startWebInterface() { if argInfo.WebURL != "" { // start web go webHelper() go web.Start(r.Server, argInfo.WebURL) } } //-------------------------------------- // HTTP Utilities //-------------------------------------- func dispatch(c Command, w http.ResponseWriter, req *http.Request, toURL func(name string) (string, bool)) error { if r.State() == raft.Leader { if response, err := r.Do(c); err != nil { return err } else { if response == nil { return etcdErr.NewError(300, "Empty response from raft", store.UndefIndex, store.UndefTerm) } event, ok := response.(*store.Event) if ok { bytes, err := json.Marshal(event) if err != nil { fmt.Println(err) } w.Header().Add("X-Etcd-Index", fmt.Sprint(event.Index)) w.Header().Add("X-Etcd-Term", fmt.Sprint(event.Term)) w.WriteHeader(http.StatusOK) w.Write(bytes) return nil } bytes, _ := response.([]byte) w.WriteHeader(http.StatusOK) w.Write(bytes) return nil } } else { leader := r.Leader() // current no leader if leader == "" { return etcdErr.NewError(300, "", store.UndefIndex, store.UndefTerm) } url, _ := toURL(leader) redirect(url, w, req) return nil } } func redirect(hostname string, w http.ResponseWriter, req *http.Request) { path := req.URL.Path url := hostname + path debugf("Redirect to %s", url) http.Redirect(w, req, url, http.StatusTemporaryRedirect) } func decodeJsonRequest(req *http.Request, data interface{}) error { decoder := json.NewDecoder(req.Body) if err := decoder.Decode(&data); err != nil && err != io.EOF { warnf("Malformed json request: %v", err) return fmt.Errorf("Malformed json request: %v", err) } return nil } func encodeJsonResponse(w http.ResponseWriter, status int, data interface{}) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) if data != nil { encoder := json.NewEncoder(w) encoder.Encode(data) } } // sanitizeURL will cleanup a host string in the format hostname:port and // attach a schema. func sanitizeURL(host string, defaultScheme string) string { // Blank URLs are fine input, just return it if len(host) == 0 { return host } p, err := url.Parse(host) if err != nil { fatal(err) } // Make sure the host is in Host:Port format _, _, err = net.SplitHostPort(host) if err != nil { fatal(err) } p = &url.URL{Host: host, Scheme: defaultScheme} return p.String() } // sanitizeListenHost cleans up the ListenHost parameter and appends a port // if necessary based on the advertised port. func sanitizeListenHost(listen string, advertised string) string { aurl, err := url.Parse(advertised) if err != nil { fatal(err) } ahost, aport, err := net.SplitHostPort(aurl.Host) if err != nil { fatal(err) } // If the listen host isn't set use the advertised host if listen == "" { listen = ahost } return net.JoinHostPort(listen, aport) } func check(err error) { if err != nil { fatal(err) } } func getNodePath(urlPath string) string { pathPrefixLen := len("/" + version + "/keys") return urlPath[pathPrefixLen:] } //-------------------------------------- // Log //-------------------------------------- var logger *log.Logger = log.New("etcd", false, log.CombinedSink(os.Stdout, "[%s] %s %-9s | %s\n", []string{"prefix", "time", "priority", "message"})) func infof(format string, v ...interface{}) { logger.Infof(format, v...) } func debugf(format string, v ...interface{}) { if verbose { logger.Debugf(format, v...) } } func debug(v ...interface{}) { if verbose { logger.Debug(v...) } } func warnf(format string, v ...interface{}) { logger.Warningf(format, v...) } func warn(v ...interface{}) { logger.Warning(v...) } func fatalf(format string, v ...interface{}) { logger.Fatalf(format, v...) } func fatal(v ...interface{}) { logger.Fatalln(v...) } //-------------------------------------- // CPU profile //-------------------------------------- func runCPUProfile() { f, err := os.Create(cpuprofile) if err != nil { fatal(err) } pprof.StartCPUProfile(f) c := make(chan os.Signal, 1) signal.Notify(c, os.Interrupt) go func() { for sig := range c { infof("captured %v, stopping profiler and exiting..", sig) pprof.StopCPUProfile() os.Exit(1) } }() } //-------------------------------------- // Testing //-------------------------------------- func directSet() { c := make(chan bool, 1000) for i := 0; i < 1000; i++ { go send(c) } for i := 0; i < 1000; i++ { <-c } } func send(c chan bool) { for i := 0; i < 10; i++ { command := &UpdateCommand{} command.Key = "foo" command.Value = "bar" command.ExpireTime = time.Unix(0, 0) r.Do(command) } c <- true }