func TestConnHashTestSuite(t *testing.T) {
	if testing.Short() {
		t.Skip("Skipping zk shard test in short mode.")
	}
	//launch zk
	sl := test.NewServiceLauncher()
	etcdOriPort, stopEtcd, err := sl.Start(test.Etcd)
	assert.NoError(t, err)
	s := new(ConnHashTestSuite)
	s.etcdOriPort = etcdOriPort
	s.etcdPort = 3333 // used for port forward
	s.forwarder = portForwarder(
		fmt.Sprintf(":%d", s.etcdPort), fmt.Sprintf(":%d", s.etcdOriPort))

	s.stopForward, err = s.forwarder()
	assert.NoError(t, err, "no error")

	// non-forward etcdCli
	etcdCli = etcd.NewClient([]string{fmt.Sprintf("http://localhost:%d", etcdOriPort)})
	// forwardable etcdCli
	etcdForwdCli = etcd.NewClient([]string{fmt.Sprintf("http://localhost:%d", s.etcdPort)})
	suite.Run(t, s)

	// clean up the forwarding
	s.stopForward <- struct{}{}
	etcdCli.Close()
	etcdForwdCli.Close()
	assert.NoError(s.T(), stopEtcd())
}

func main() {

	// Setup
	servers := []string{"http://localhost:4001"}
	log.Printf("Creating etcd client pointing to %v", servers)
	etcdClient := etcd.NewClient(servers)
	machineList := []string{"machine"}

	reg := registry.MakeEtcdRegistry(etcdClient, machineList)

	apiserver := apiserver.New(map[string]apiserver.RESTStorage{
		"pods": registry.MakePodRegistryStorage(reg, &client.FakeContainerInfo{}, registry.MakeRoundRobinScheduler(machineList)),
		"replicationControllers": registry.MakeControllerRegistryStorage(reg),
	}, "/api/v1beta1")
	server := httptest.NewServer(apiserver)

	controllerManager := registry.MakeReplicationManager(etcd.NewClient(servers),
		client.Client{
			Host: server.URL,
		})

	go controllerManager.Synchronize()
	go controllerManager.WatchControllers()

	// Ok. we're good to go.
	log.Printf("API Server started on %s", server.URL)
	// Wait for the synchronization threads to come up.
	time.Sleep(time.Second * 10)

	kubeClient := client.Client{
		Host: server.URL,
	}
	data, err := ioutil.ReadFile("api/examples/controller.json")
	if err != nil {
		log.Fatalf("Unexpected error: %#v", err)
	}
	var controllerRequest api.ReplicationController
	if err = json.Unmarshal(data, &controllerRequest); err != nil {
		log.Fatalf("Unexpected error: %#v", err)
	}

	if _, err = kubeClient.CreateReplicationController(controllerRequest); err != nil {
		log.Fatalf("Unexpected error: %#v", err)
	}
	// Give the controllers some time to actually create the pods
	time.Sleep(time.Second * 10)

	// Validate that they're truly up.
	pods, err := kubeClient.ListPods(nil)
	if err != nil || len(pods.Items) != 2 {
		log.Fatal("FAILED")
	}
	log.Printf("OK")
}

func (r *Reconfigurator) Init(host, port string) error {
	r.configs = make(map[string]*Config)

	if host == "" {
		return errors.New("Invalid host: " + host)
	} else {
		r.etcdHost = host
	}

	if port == "" {
		return errors.New("Invalid port: " + port)
	} else {
		r.etcdPort = port
	}

	// make sure host has expected prefix
	etcdHost := host
	if !strings.HasPrefix(host, "http://") {
		etcdHost = "http://" + host
	}

	client := etcd.NewClient([]string{etcdHost + ":" + port})

	if client == nil {
		return errors.New("Unable to created ETCD client")
	} else {
		r.client = client
		return nil
	}
}

func main() {
	host := getopt("HOST", "127.0.0.1")

	etcdPort := getopt("ETCD_PORT", "4001")
	etcdPath := getopt("ETCD_PATH", "/deis/cache")

	externalPort := getopt("EXTERNAL_PORT", "6379")

	client := etcd.NewClient([]string{"http://" + host + ":" + etcdPort})

	var maxmemory string
	result, err := client.Get("/deis/cache/maxmemory", false, false)
	if err != nil {
		if e, ok := err.(*etcd.EtcdError); ok && e.ErrorCode == ectdKeyNotFound {
			maxmemory = defaultMemory
		} else {
			log.Fatalln(err)
		}
	} else {
		maxmemory = result.Node.Key
	}
	replaceMaxmemoryInConfig(maxmemory)

	go launchRedis()

	go publishService(client, host, etcdPath, externalPort, uint64(ttl.Seconds()))

	// Wait for terminating signal
	exitChan := make(chan os.Signal, 2)
	signal.Notify(exitChan, syscall.SIGTERM, syscall.SIGINT)
	<-exitChan
}

// Main method.
func main() {
	conf := config.New()

	help, version := false, false
	flag.BoolVar(&help, "help", false, "Prints command line options and exit.")
	flag.BoolVar(&version, "version", false, "Prints the etcdsh version and exit.")
	flag.StringVar(&conf.Machine, "machine", conf.Machine, "Connect to this etcd server.")
	flag.StringVar(&conf.PS1, "ps1", conf.PS1, "First prompt format")
	flag.StringVar(&conf.PS2, "ps2", conf.PS2, "Second prompt format")
	flag.BoolVar(&conf.Colors, "colors", conf.Colors, "Use colors in display.")
	flag.Parse()

	if help {
		printHelp()
		os.Exit(0)
	}
	if version {
		printVersion()
		os.Exit(0)
	}

	fmt.Printf("Connecting to %s\n", conf.Machine)
	client := etcd.NewClient([]string{conf.Machine})

	controller := handlers.NewController(conf, client, os.Stdout, os.Stderr, os.Stdin)
	controller.Add(handlers.NewLsHandler(controller))
	controller.Add(handlers.NewSetHandler(controller))
	controller.Add(handlers.NewHelpHandler(controller))
	controller.Add(handlers.NewCdHandler(controller))
	controller.Add(handlers.NewGetHandler(controller))
	os.Exit(controller.Start())
}

func main() {
	name := os.Getenv("VIRTUAL_HOST")
	site := NewSite(name)
	client := etcd.NewClient([]string{os.Getenv("ETCD_URL")})
	go AnnounceSite(name, site, client)
	select {}
}

func NewClientEtcd(uqProtocol string, etcdAddr []string, etcdKey string) (*Client, error) {
	client := new(Client)
	if len(etcdAddr) > 0 {
		etcdClient := etcd.NewClient(etcdAddr)
		var uqConn Conn
		var err error
		if uqProtocol == ProtoRedis {
			uqConn, err = newConnRedisEtcd(etcdClient, etcdKey)
			if err != nil {
				return nil, err
			}
		} else if uqProtocol == ProtoMc {
			uqConn, err = newConnMcEtcd(etcdClient, etcdKey)
			if err != nil {
				return nil, err
			}
		} else if uqProtocol == ProtoHttp {
			uqConn, err = newConnHttpEtcd(etcdClient, etcdKey)
			if err != nil {
				return nil, err
			}
		}
		client.uqConn = uqConn
	} else {
		return nil, errors.New("etcd address nil")
	}
	return client, nil
}

func newKubernetesMaster(scheduler *scheduler.KubernetesScheduler, c *master.Config) *kubernetesMaster {
	var m *kubernetesMaster

	if len(c.EtcdServers) > 0 {
		etcdClient := etcd.NewClient(c.EtcdServers)
		minionRegistry := registry.MakeMinionRegistry(c.Minions)
		m = &kubernetesMaster{
			podRegistry:        scheduler,
			controllerRegistry: registry.MakeEtcdRegistry(etcdClient, minionRegistry),
			serviceRegistry:    registry.MakeEtcdRegistry(etcdClient, minionRegistry),
			minionRegistry:     minionRegistry,
			client:             c.Client,
		}
		m.init(scheduler, c.Cloud, c.PodInfoGetter)
	} else {
		m = &kubernetesMaster{
			podRegistry:        scheduler,
			controllerRegistry: registry.MakeMemoryRegistry(),
			serviceRegistry:    registry.MakeMemoryRegistry(),
			minionRegistry:     registry.MakeMinionRegistry(c.Minions),
			client:             c.Client,
		}
		m.init(scheduler, c.Cloud, c.PodInfoGetter)
	}
	return m
}

// Starts background goroutines. If file, manifest_url, or address are empty,
// they are not watched. Never returns.
func (kl *Kubelet) RunKubelet(file, manifest_url, etcd_servers, address string, port uint) {
	fileChannel := make(chan api.ContainerManifest)
	etcdChannel := make(chan []api.ContainerManifest)
	httpChannel := make(chan api.ContainerManifest)
	serverChannel := make(chan api.ContainerManifest)

	go util.Forever(func() { kl.WatchFile(file, fileChannel) }, 20*time.Second)
	if manifest_url != "" {
		go util.Forever(func() { kl.WatchHTTP(manifest_url, httpChannel) }, 20*time.Second)
	}
	if etcd_servers != "" {
		servers := []string{etcd_servers}
		log.Printf("Creating etcd client pointing to %v", servers)
		kl.Client = etcd.NewClient(servers)
		go util.Forever(func() { kl.SyncAndSetupEtcdWatch(etcdChannel) }, 20*time.Second)
	}
	if address != "" {
		log.Printf("Starting to listen on %s:%d", address, port)
		handler := KubeletServer{
			Kubelet:       kl,
			UpdateChannel: serverChannel,
		}
		s := &http.Server{
			// TODO: This is broken if address is an ipv6 address.
			Addr:           fmt.Sprintf("%s:%d", address, port),
			Handler:        &handler,
			ReadTimeout:    10 * time.Second,
			WriteTimeout:   10 * time.Second,
			MaxHeaderBytes: 1 << 20,
		}
		go util.Forever(func() { s.ListenAndServe() }, 0)
	}
	kl.RunSyncLoop(etcdChannel, fileChannel, serverChannel, httpChannel, kl)
}

// EtcdClient creates an etcd client based on the provided config.
func EtcdClient(etcdClientInfo configapi.EtcdConnectionInfo) (*etcdclient.Client, error) {
	tlsConfig, err := client.TLSConfigFor(&client.Config{
		TLSClientConfig: client.TLSClientConfig{
			CertFile: etcdClientInfo.ClientCert.CertFile,
			KeyFile:  etcdClientInfo.ClientCert.KeyFile,
			CAFile:   etcdClientInfo.CA,
		},
	})
	if err != nil {
		return nil, err
	}

	transport := knet.SetTransportDefaults(&http.Transport{
		TLSClientConfig: tlsConfig,
		Dial: (&net.Dialer{
			// default from http.DefaultTransport
			Timeout: 30 * time.Second,
			// Lower the keep alive for connections.
			KeepAlive: 1 * time.Second,
		}).Dial,
		// Because watches are very bursty, defends against long delays in watch reconnections.
		MaxIdleConnsPerHost: 500,
	})

	etcdClient := etcdclient.NewClient(etcdClientInfo.URLs)
	etcdClient.SetTransport(transport)
	etcdClient.CheckRetry = NeverRetryOnFailure
	return etcdClient, nil
}

// ExampleNestedStruct creates a Etcd directory using a nested Go struct and then gets the directory as JSON.
func Example_nestedStruct() {
	verbose := flag.Bool("verbose", false, "Verbose")
	node := flag.String("node", "", "Etcd node")
	port := flag.String("port", "", "Etcd port")
	flag.Parse()

	// Define nested structure.
	g := Group{
		Name: "staff",
		Users: []User{
			User{
				Name:      "jdoe",
				Age:       25,
				Male:      true,
				FirstName: "John",
				LastName:  "Doe",
			},
			User{
				Name:      "lnemoy",
				Age:       62,
				Male:      true,
				FirstName: "Leonard",
				LastName:  "Nimoy",
			},
		},
	}

	// Connect to Etcd.
	conn := getEnv()
	if node == nil && port == nil {
		conn = []string{fmt.Sprintf("http://%v:%v", *node, *port)}
	}

	if *verbose {
		log.Printf("Connecting to: %s", conn)
	}
	client := etcd.NewClient(conn)

	// Create directory structure based on struct.
	err := etcdmap.Create(client, "/example", reflect.ValueOf(g))
	if err != nil {
		log.Fatal(err.Error())
	}

	// Get directory structure from Etcd.
	res, err := client.Get("/example", true, true)
	if err != nil {
		log.Fatal(err.Error())
	}

	j, err2 := etcdmap.JSON(res.Node)
	if err2 != nil {
		log.Fatal(err2.Error())
	}

	fmt.Println(string(j))

	// Output:
	//{"id":"staff","users":{"0":{"age":"25","first_name":"John","id":"jdoe","last_name":"Doe","male":"true"},"1":{"age":"62","first_name":"Leonard","id":"lnemoy","last_name":"Nimoy","male":"true"}}}
}

// etcdRecordCheckin takes a server struct and stores it in Etcd as a
// json value
func (s *Server) etcdRecordCheckin() {
	if s.Hostname == "" {
		log.Printf("[WARNING] Unable to checkin server with Ip Address %s due to no hostname (can't resolve and wasn't supplied in JSON).", s.IpAddress)
		return
	}
	if *Verbose {
		log.Printf("connecting to: %s%s", clientConnectionSetup(), Endpoint)
	}
	client := etcd.NewClient(clientConnectionSetup())
	// first ensure the Endpoint is registered
	_, err := client.Get(Endpoint, false, false)
	if err != nil {
		_, err := client.SetDir(Endpoint, 0)
		if err != nil {
			log.Printf("[ERROR] Etcd SetDir Error: %s", err)
			return
		}
	}
	// generate json from the server struct
	json_value, err := json.Marshal(s)

	if err != nil {
		log.Printf("[ERROR] Json Marshal Error: %s", err)
		return
	}
	// now we set the value
	client.Set(Endpoint+"/"+s.Hostname, string(json_value), 0)
}

func (w *watcher) Init(etcdURLS string) {
	w.client = etcd.NewClient(strings.Split(etcdURLS, ","))

	if len(w.etcdMachines) > 0 {
		w.client.SetCluster(w.etcdMachines)
	}
}

func NewClient(key string) *Client {
	etcd := etcd.NewClient([]string{etcdMachine})

	etcd.SyncCluster()

	return &Client{Key: key, etcd: etcd}
}

func NewEtcd(prefixKey string, config *EtcdConfig) (*Etcd, error) {
	var (
		client *etcd.Client
		err    error
	)

	if config.CertFile != "" && config.KeyFile != "" {
		if client, err = etcd.NewTLSClient(config.Machines, config.CertFile, config.KeyFile, config.CaFile); err != nil {
			Logger.Error("Failed to connect to Etcd. Error: %+v.", err)
			return nil, err
		}
	} else {
		client = etcd.NewClient(config.Machines)
	}

	// Set the default value if not provided.
	if config.EtcdConsistency == "" {
		config.EtcdConsistency = etcd.STRONG_CONSISTENCY
	}

	if err = client.SetConsistency(config.EtcdConsistency); err != nil {
		Logger.Error("Failed to set Etcd consitency. Error: %+v.", err)
		return nil, err
	}

	return &Etcd{client: client, prefixKey: prefixKey}, nil
}

func (f *Factory) New(uri *url.URL) bridge.RegistryAdapter {
	urls := make([]string, 0)
	if uri.Host != "" {
		urls = append(urls, "http://"+uri.Host)
	}
	return &EtcdAdapter{client: etcd.NewClient(urls), path: uri.Path}
}

// HomeHandler is the main home screen
func HomeHandler(w http.ResponseWriter, r *http.Request) {
	c, err := goship.ParseETCD(etcd.NewClient([]string{*ETCDServer}))
	if err != nil {
		log.Printf("Failed to Parse to ETCD data %s", err)
		http.Error(w, err.Error(), http.StatusInternalServerError)
	}
	u, err := getUser(r)
	if err != nil {
		log.Println("Failed to get User! ")
		http.Error(w, err.Error(), http.StatusUnauthorized)
	}
	t, err := template.New("index.html").ParseFiles("templates/index.html", "templates/base.html")
	if err != nil {
		log.Printf("Failed to parse template: %s", err)
		http.Error(w, err.Error(), http.StatusInternalServerError)
	}
	c.Projects = removeUnauthorizedProjects(c.Projects, r, u)

	sort.Sort(ByName(c.Projects))

	// apply each plugin
	for _, pl := range plugin.Plugins {
		err := pl.Apply(c)
		if err != nil {
			log.Printf("Failed to apply plugin: %s", err)
			http.Error(w, err.Error(), http.StatusInternalServerError)
		}
	}
	js, css := getAssetsTemplates()
	gt := os.Getenv(gitHubAPITokenEnvVar)
	pt := c.Pivotal.Token

	t.ExecuteTemplate(w, "base", map[string]interface{}{"Javascript": js, "Stylesheet": css, "Projects": c.Projects, "User": u, "Page": "home", "ConfirmDeployFlag": *confirmDeployFlag, "GithubToken": gt, "PivotalToken": pt})
}

func NewEtcdStore(uri *url.URL) ConfigStore {
	urls := make([]string, 0)
	if uri.Host != "" {
		urls = append(urls, "http://"+uri.Host)
	}
	return &EtcdStore{client: etcd.NewClient(urls)}
}

func Client() *etcd.Client {
	clientSingletonO.Do(func() {
		host := "http://localhost:4001"
		if len(os.Getenv("ETCD_HOST")) != 0 {
			host = os.Getenv("ETCD_HOST")
		}

		cacert := os.Getenv("ETCD_CACERT")
		tlskey := os.Getenv("ETCD_TLS_KEY")
		tlscert := os.Getenv("ETCD_TLS_CERT")
		if len(cacert) != 0 && len(tlskey) != 0 && len(tlscert) != 0 {
			if !strings.Contains(host, "https://") {
				host = strings.Replace(host, "http", "https", 1)
			}
			c, err := etcd.NewTLSClient([]string{host}, tlscert, tlskey, cacert)
			if err != nil {
				panic(err)
			}
			clientSingleton = c
		} else {
			clientSingleton = etcd.NewClient([]string{host})
		}
	})
	return clientSingleton
}

func main() {
	// TODO: Add healthz and use it for flannel
	flags.Parse(os.Args)
	if *networkPath == "" {
		log.Printf("Please specify a --network-config argument.")
		return
	}
	if _, err := os.Stat(*networkPath); err != nil {
		log.Fatalf("Did not find network config file %v", *networkPath)
	}
	buff, err := ioutil.ReadFile(*networkPath)
	if err != nil {
		log.Fatalf("Unable to read network configuration: %v", err)
	}

	client := etcd.NewClient([]string{*etcdServer})
	networkBlob := string(buff)
	configPath := fmt.Sprintf("%v/%v", *etcdPrefix, "config")
	if _, err := client.Set(configPath, networkBlob, uint64(0)); err != nil {
		log.Fatalf("Unable to create network configuration: %+v", err)
	}
	log.Printf("Created network %v", networkBlob)

	log.Fatalf(fmt.Sprintf("%v", http.ListenAndServe(fmt.Sprintf(":%v", *healthzPort), nil)))
}