// DefaultConfig is used to return a sane default configuration
func DefaultConfig() *Config {
	hostname, err := os.Hostname()
	if err != nil {
		panic(err)
	}

	conf := &Config{
		Datacenter:        DefaultDC,
		NodeName:          hostname,
		RPCAddr:           DefaultRPCAddr,
		RaftConfig:        raft.DefaultConfig(),
		SerfLANConfig:     serf.DefaultConfig(),
		SerfWANConfig:     serf.DefaultConfig(),
		ReconcileInterval: 60 * time.Second,
		ProtocolVersion:   ProtocolVersionMax,
	}

	// Increase our reap interval to 3 days instead of 24h.
	conf.SerfLANConfig.ReconnectTimeout = 3 * 24 * time.Hour
	conf.SerfWANConfig.ReconnectTimeout = 3 * 24 * time.Hour

	// WAN Serf should use the WAN timing, since we are using it
	// to communicate between DC's
	conf.SerfWANConfig.MemberlistConfig = memberlist.DefaultWANConfig()

	// Ensure we don't have port conflicts
	conf.SerfLANConfig.MemberlistConfig.BindPort = DefaultLANSerfPort
	conf.SerfWANConfig.MemberlistConfig.BindPort = DefaultWANSerfPort

	// Disable shutdown on removal
	conf.RaftConfig.ShutdownOnRemove = false

	return conf
}

// DefaultConfig is used to return a sane default configuration
func DefaultConfig() *Config {
	hostname, err := os.Hostname()
	if err != nil {
		panic(err)
	}

	conf := &Config{
		Datacenter:               DefaultDC,
		NodeName:                 hostname,
		RPCAddr:                  DefaultRPCAddr,
		RaftConfig:               raft.DefaultConfig(),
		SerfLANConfig:            serf.DefaultConfig(),
		SerfWANConfig:            serf.DefaultConfig(),
		ReconcileInterval:        60 * time.Second,
		ProtocolVersion:          ProtocolVersion2Compatible,
		ACLTTL:                   30 * time.Second,
		ACLDefaultPolicy:         "allow",
		ACLDownPolicy:            "extend-cache",
		ACLReplicationInterval:   30 * time.Second,
		ACLReplicationApplyLimit: 100, // ops / sec
		TombstoneTTL:             15 * time.Minute,
		TombstoneTTLGranularity:  30 * time.Second,
		SessionTTLMin:            10 * time.Second,
		DisableCoordinates:       false,

		// These are tuned to provide a total throughput of 128 updates
		// per second. If you update these, you should update the client-
		// side SyncCoordinateRateTarget parameter accordingly.
		CoordinateUpdatePeriod:     5 * time.Second,
		CoordinateUpdateBatchSize:  128,
		CoordinateUpdateMaxBatches: 5,

		// Hold an RPC for up to 5 seconds by default
		RPCHoldTimeout: 5 * time.Second,
	}

	// Increase our reap interval to 3 days instead of 24h.
	conf.SerfLANConfig.ReconnectTimeout = 3 * 24 * time.Hour
	conf.SerfWANConfig.ReconnectTimeout = 3 * 24 * time.Hour

	// WAN Serf should use the WAN timing, since we are using it
	// to communicate between DC's
	conf.SerfWANConfig.MemberlistConfig = memberlist.DefaultWANConfig()

	// Ensure we don't have port conflicts
	conf.SerfLANConfig.MemberlistConfig.BindPort = DefaultLANSerfPort
	conf.SerfWANConfig.MemberlistConfig.BindPort = DefaultWANSerfPort

	// Enable interoperability with unversioned Raft library, and don't
	// start using new ID-based features yet.
	conf.RaftConfig.ProtocolVersion = 1

	// Disable shutdown on removal
	conf.RaftConfig.ShutdownOnRemove = false

	return conf
}

// DefaultConfig is used to return a sane default configuration
func DefaultConfig() *Config {
	hostname, err := os.Hostname()
	if err != nil {
		panic(err)
	}

	conf := &Config{
		Datacenter:              DefaultDC,
		NodeName:                hostname,
		RPCAddr:                 DefaultRPCAddr,
		RaftConfig:              raft.DefaultConfig(),
		SerfLANConfig:           serf.DefaultConfig(),
		SerfWANConfig:           serf.DefaultConfig(),
		ReconcileInterval:       60 * time.Second,
		ProtocolVersion:         ProtocolVersion2Compatible,
		ACLTTL:                  30 * time.Second,
		ACLDefaultPolicy:        "allow",
		ACLDownPolicy:           "extend-cache",
		TombstoneTTL:            15 * time.Minute,
		TombstoneTTLGranularity: 30 * time.Second,
		SessionTTLMin:           10 * time.Second,
		DisableCoordinates:      false,

		// These are tuned to provide a total throughput of 128 updates
		// per second. If you update these, you should update the client-
		// side SyncCoordinateRateTarget parameter accordingly.
		CoordinateUpdatePeriod:     5 * time.Second,
		CoordinateUpdateBatchSize:  128,
		CoordinateUpdateMaxBatches: 5,
	}

	// Increase our reap interval to 3 days instead of 24h.
	conf.SerfLANConfig.ReconnectTimeout = 3 * 24 * time.Hour
	conf.SerfWANConfig.ReconnectTimeout = 3 * 24 * time.Hour

	// WAN Serf should use the WAN timing, since we are using it
	// to communicate between DC's
	conf.SerfWANConfig.MemberlistConfig = memberlist.DefaultWANConfig()

	// Turn LAN Serf to run globally
	conf.SerfLANConfig.MemberlistConfig = memberlist.DefaultWANConfig()

	// Ensure we don't have port conflicts
	conf.SerfLANConfig.MemberlistConfig.BindPort = DefaultLANSerfPort
	conf.SerfWANConfig.MemberlistConfig.BindPort = DefaultWANSerfPort

	// Disable shutdown on removal
	conf.RaftConfig.ShutdownOnRemove = false

	// Make Raft more WAN friendly
	conf.RaftConfig.HeartbeatTimeout = 5000 * time.Millisecond
	conf.RaftConfig.ElectionTimeout = 5000 * time.Millisecond
	conf.RaftConfig.CommitTimeout = 100 * time.Millisecond
	conf.RaftConfig.LeaderLeaseTimeout = 2500 * time.Millisecond

	return conf
}

func TestAgentTagsFile(t *testing.T) {
	tags := map[string]string{
		"role":       "webserver",
		"datacenter": "us-east",
	}

	td, err := ioutil.TempDir("", "serf")
	if err != nil {
		t.Fatalf("err: %s", err)
	}
	defer os.RemoveAll(td)

	agentConfig := DefaultConfig()
	agentConfig.TagsFile = filepath.Join(td, "tags.json")

	a1 := testAgentWithConfig(agentConfig, serf.DefaultConfig(), nil)

	if err := a1.Start(); err != nil {
		t.Fatalf("err: %s", err)
	}
	defer a1.Shutdown()
	defer a1.Leave()

	testutil.Yield()

	err = a1.SetTags(tags)

	if err != nil {
		t.Fatalf("err: %s", err)
	}

	testutil.Yield()

	a2 := testAgentWithConfig(agentConfig, serf.DefaultConfig(), nil)

	if err := a2.Start(); err != nil {
		t.Fatalf("err: %s", err)
	}
	defer a2.Shutdown()
	defer a2.Leave()

	testutil.Yield()

	m := a2.Serf().LocalMember()

	if !reflect.DeepEqual(m.Tags, tags) {
		t.Fatalf("tags not restored: %#v", m.Tags)
	}
}

func Create() (*Agent, error) {
	conf := serf.DefaultConfig()
	conf.Init()
	conf.NodeName = os.Getenv("HOST")
	conf.Tags["DOCKER_HOST"] = Os.Getenv("DOCKER_HOST")

	logOutput := log.StandardLogger().Out

	// Setup the underlying loggers
	conf.MemberlistConfig.LogOutput = logOutput
	conf.LogOutput = logOutput

	// Create a channel to listen for events from Serf
	eventCh := make(chan serf.Event, 64)
	conf.EventCh = eventCh

	// support only LAN configuration at the moment
	conf.MemberlistConfig = memberlist.DefaultLANConfig()
	conf.MemberlistConfig.BindAddr = "0.0.0.0"
	conf.MemberlistConfig.BindPort = 3388

	// Setup the agent
	agent := &Agent{
		conf:       conf,
		eventCh:    eventCh,
		isManager:  false,
		shutdownCh: make(chan struct{}),
	}

	return agent, nil
}

func (a *Agent) Start() error {
	conf := serf.DefaultConfig()
	conf.Init()
	conf.Tags["role"] = "agent"
	conf.Tags["info"] = a.getInfo()

	bindAddr, err := a.config.GetIpByInterface()
	if err != nil {
		log.Fatal(err)
	}

	conf.NodeName = a.config.Name

	conf.MemberlistConfig.BindAddr = bindAddr
	conf.EventCh = a.eventCh

	serf, err := serf.Create(conf)
	if err != nil {
		return err
	}

	a.serf = serf

	go a.handleEvents()

	return nil
}

func (n *Node) createSerfAgent(iface string) {

	serfConfig := serf.DefaultConfig()

	serfConfig.Tags = n.tags

	serfConfig.MemberlistConfig.BindAddr = iface
	serfConfig.MemberlistConfig.BindPort = getRandomPort(iface)

	serfConfig.NodeName = fmt.Sprintf("%[email protected]%s", n.UUID, iface)
	serfConfig.Tags = n.tags
	serfConfig.LogOutput = ioutil.Discard
	//	serfConfig.MemberlistConfig.GossipInterval = 5 * time.Millisecond
	//	serfConfig.MemberlistConfig.ProbeInterval = 50 * time.Millisecond
	//	serfConfig.MemberlistConfig.ProbeTimeout = 25 * time.Millisecond
	//	serfConfig.MemberlistConfig.SuspicionMult = 1
	serfConfig.Init()
	agentConfig := agent.DefaultConfig()
	agentConfig.Tags = n.tags
	agentConfig.LogLevel = "INFO"
	agt, err := agent.Create(agentConfig, serfConfig, ioutil.Discard)
	//	agt, err := agent.Create(agentConfig, serfConfig, n.cfg.Logger())

	if n.handleErr(err) {
		eventHandler := newEventHandler()
		n.join.Join(eventHandler.join.WhereNot(n.isSelf))
		n.leave.Join(eventHandler.leave.WhereNot(n.isSelf).Transform(toLeaveEvent()))
		n.query.Join(eventHandler.query.Transform(toQueryEvent(iface)))
		agt.RegisterEventHandler(eventHandler)
		n.agents[iface] = agt
		n.logger.Println("Agent Created On Port", agt.SerfConfig().MemberlistConfig.BindPort)
	} else {
		n.logger.Println("Failed to create Agent")
	}
}

func main() {
	InitLogger(logrus.DebugLevel.String(), "yws1")

	serfConfig := serf.DefaultConfig()
	serfConfig.NodeName = "yws1"
	serfConfig.MemberlistConfig = memberlist.DefaultLANConfig()

	serfConfig.MemberlistConfig.BindAddr = "127.0.0.1"
	serfConfig.MemberlistConfig.BindPort = 7373
	serfConfig.MemberlistConfig.AdvertiseAddr = "127.0.0.1"
	serfConfig.MemberlistConfig.AdvertisePort = 5000

	eventCh := make(chan serf.Event, 64)
	serfConfig.EventCh = eventCh
	serfConfig.LogOutput = ioutil.Discard
	serfConfig.MemberlistConfig.LogOutput = ioutil.Discard

	s, err := serf.Create(serfConfig)
	if err != nil {
		fmt.Println(err)
	}

	go func() {
		serfShutdownCh := s.ShutdownCh()
		log.Info("agent: Listen for events")
		for {
			select {
			case e := <-eventCh:
				log.WithFields(logrus.Fields{
					"event": e.String(),
				}).Debug("agent: Received event")

				// Log all member events
				if failed, ok := e.(serf.MemberEvent); ok {
					for _, member := range failed.Members {
						log.WithFields(logrus.Fields{
							"node":   "yang",
							"member": member.Name,
							"event":  e.EventType(),
						}).Debug("agent: Member event")
					}
				}

				if e.EventType() == serf.EventQuery {
					query := e.(*serf.Query)
					fmt.Println(query)
				}

			case <-serfShutdownCh:
				log.Warn("agent: Serf shutdown detected, quitting")
				return
			}
		}
	}()

	fmt.Println(s)

	wait := make(chan int)
	<-wait
}

func (d *driver) serfInit() error {
	var err error

	config := serf.DefaultConfig()
	config.Init()
	config.MemberlistConfig.BindAddr = d.bindAddress

	d.eventCh = make(chan serf.Event, 4)
	config.EventCh = d.eventCh
	config.UserCoalescePeriod = 1 * time.Second
	config.UserQuiescentPeriod = 50 * time.Millisecond

	config.LogOutput = &logWriter{}
	config.MemberlistConfig.LogOutput = config.LogOutput

	s, err := serf.Create(config)
	if err != nil {
		return fmt.Errorf("failed to create cluster node: %v", err)
	}
	defer func() {
		if err != nil {
			s.Shutdown()
		}
	}()

	d.serfInstance = s

	d.notifyCh = make(chan ovNotify)
	d.exitCh = make(chan chan struct{})

	go d.startSerfLoop(d.eventCh, d.notifyCh, d.exitCh)
	return nil
}

// setupAgent is used to create the agent we use
func (c *Command) setupAgent(config *Config, logOutput io.Writer) *Agent {
	bindIP, bindPort, err := config.AddrParts(config.BindAddr)
	if err != nil {
		c.Ui.Error(fmt.Sprintf("Invalid bind address: %s", err))
		return nil
	}

	var advertiseIP string
	var advertisePort int
	if config.AdvertiseAddr != "" {
		advertiseIP, advertisePort, err = config.AddrParts(config.AdvertiseAddr)
		if err != nil {
			c.Ui.Error(fmt.Sprintf("Invalid advertise address: %s", err))
			return nil
		}
	}

	encryptKey, err := config.EncryptBytes()
	if err != nil {
		c.Ui.Error(fmt.Sprintf("Invalid encryption key: %s", err))
		return nil
	}

	serfConfig := serf.DefaultConfig()
	switch config.Profile {
	case "lan":
		serfConfig.MemberlistConfig = memberlist.DefaultLANConfig()
	case "wan":
		serfConfig.MemberlistConfig = memberlist.DefaultWANConfig()
	case "local":
		serfConfig.MemberlistConfig = memberlist.DefaultLocalConfig()
	default:
		c.Ui.Error(fmt.Sprintf("Unknown profile: %s", config.Profile))
		return nil
	}

	serfConfig.MemberlistConfig.BindAddr = bindIP
	serfConfig.MemberlistConfig.BindPort = bindPort
	serfConfig.MemberlistConfig.AdvertiseAddr = advertiseIP
	serfConfig.MemberlistConfig.AdvertisePort = advertisePort
	serfConfig.MemberlistConfig.SecretKey = encryptKey
	serfConfig.NodeName = config.NodeName
	serfConfig.Tags = config.Tags
	serfConfig.SnapshotPath = config.SnapshotPath
	serfConfig.ProtocolVersion = uint8(config.Protocol)
	serfConfig.CoalescePeriod = 3 * time.Second
	serfConfig.QuiescentPeriod = time.Second
	serfConfig.UserCoalescePeriod = 3 * time.Second
	serfConfig.UserQuiescentPeriod = time.Second

	// Start Serf
	c.Ui.Output("Starting Serf agent...")
	agent, err := Create(serfConfig, logOutput)
	if err != nil {
		c.Ui.Error(fmt.Sprintf("Failed to start the Serf agent: %v", err))
		return nil
	}
	return agent
}

// DefaultNodeConfig returns a reasonable default to use.
func DefaultNodeConfig() *NodeConfig {
	return &NodeConfig{
		Replicas:       2,
		RpcPort:        11211,
		SerfConfig:     serf.DefaultConfig(),
		MaxMemoryUsage: 1 << 30, // 1 GB
	}
}

// DefaultConfig returns the default configuration
func DefaultConfig() *Config {
	hostname, err := os.Hostname()
	if err != nil {
		panic(err)
	}

	c := &Config{
		Region:                 DefaultRegion,
		Datacenter:             DefaultDC,
		NodeName:               hostname,
		ProtocolVersion:        ProtocolVersionMax,
		RaftConfig:             raft.DefaultConfig(),
		RaftTimeout:            10 * time.Second,
		LogOutput:              os.Stderr,
		RPCAddr:                DefaultRPCAddr,
		SerfConfig:             serf.DefaultConfig(),
		NumSchedulers:          1,
		ReconcileInterval:      60 * time.Second,
		EvalGCInterval:         5 * time.Minute,
		EvalGCThreshold:        1 * time.Hour,
		JobGCInterval:          5 * time.Minute,
		JobGCThreshold:         4 * time.Hour,
		NodeGCInterval:         5 * time.Minute,
		NodeGCThreshold:        24 * time.Hour,
		EvalNackTimeout:        60 * time.Second,
		EvalDeliveryLimit:      3,
		MinHeartbeatTTL:        10 * time.Second,
		MaxHeartbeatsPerSecond: 50.0,
		HeartbeatGrace:         10 * time.Second,
		FailoverHeartbeatTTL:   300 * time.Second,
		ConsulConfig:           config.DefaultConsulConfig(),
		VaultConfig:            config.DefaultVaultConfig(),
		RPCHoldTimeout:         5 * time.Second,
		TLSConfig:              &config.TLSConfig{},
	}

	// Enable all known schedulers by default
	c.EnabledSchedulers = make([]string, 0, len(scheduler.BuiltinSchedulers))
	for name := range scheduler.BuiltinSchedulers {
		c.EnabledSchedulers = append(c.EnabledSchedulers, name)
	}
	c.EnabledSchedulers = append(c.EnabledSchedulers, structs.JobTypeCore)

	// Default the number of schedulers to match the coores
	c.NumSchedulers = runtime.NumCPU()

	// Increase our reap interval to 3 days instead of 24h.
	c.SerfConfig.ReconnectTimeout = 3 * 24 * time.Hour

	// Serf should use the WAN timing, since we are using it
	// to communicate between DC's
	c.SerfConfig.MemberlistConfig = memberlist.DefaultWANConfig()
	c.SerfConfig.MemberlistConfig.BindPort = DefaultSerfPort

	// Disable shutdown on removal
	c.RaftConfig.ShutdownOnRemove = false
	return c
}

func serfInit() *serf.Serf {
	conf := serf.DefaultConfig()
	serfin, err := serf.Create(conf)
	if err != nil {
		log.Fatal(err)
	}

	return serfin
}

func TestAgentKeyringFile_BadOptions(t *testing.T) {
	agentConfig := DefaultConfig()
	agentConfig.KeyringFile = "/some/path"
	agentConfig.EncryptKey = "pL4owv4IE1x+ZXCyd5vLLg=="

	_, err := Create(agentConfig, serf.DefaultConfig(), nil)
	if err == nil || !strings.Contains(err.Error(), "not allowed") {
		t.Fatalf("err: %s", err)
	}
}

// DefaultConfig returns a Config with sane default values.
func DefaultConfig() Config {
	return Config{
		Partitions: 512,
		Redundancy: 3,
		SerfConfig: serf.DefaultConfig(),
		Events:     make(chan Event),
		LogOutput:  os.Stderr,
		LogLevel:   2,
	}
}

func TestAgentTagsFile_BadOptions(t *testing.T) {
	agentConfig := DefaultConfig()
	agentConfig.TagsFile = "/some/path"
	agentConfig.Tags = map[string]string{
		"tag1": "val1",
	}

	_, err := Create(agentConfig, serf.DefaultConfig(), nil)
	if err == nil || !strings.Contains(err.Error(), "not allowed") {
		t.Fatalf("err: %s", err)
	}
}

func testAgent() *Agent {
	config := serf.DefaultConfig()
	config.MemberlistConfig.BindAddr = testutil.GetBindAddr().String()
	config.NodeName = config.MemberlistConfig.BindAddr

	agent := &Agent{
		RPCAddr:    getRPCAddr(),
		SerfConfig: config,
	}

	return agent
}

func (s *Server) setupSerf() (*serf.Serf, error) {
	conf := serf.DefaultConfig()

	// Generate NodeName if missing
	id, err := uuid.UUID4()
	if err != nil {
		return nil, err
	}

	// Get SSH server port
	port := s.sshServer.Addr.Port

	// Initialize serf
	conf.Init()
	// s.logger.Info("local port", "port", conf.MemberlistConfig.BindPort)

	conf.NodeName = s.config.NodeName
	conf.MemberlistConfig.BindAddr = s.config.GossipBindAddr
	conf.MemberlistConfig.BindPort = s.config.GossipBindPort
	conf.MemberlistConfig.AdvertiseAddr = s.config.GossipAdvertiseAddr
	conf.MemberlistConfig.AdvertisePort = s.config.GossipAdvertisePort
	s.logger.Info("Gossip", "BindAddr", conf.MemberlistConfig.BindAddr, "BindPort", conf.MemberlistConfig.BindPort, "AdvertiseAddr", conf.MemberlistConfig.AdvertiseAddr, "AdvertisePort", conf.MemberlistConfig.AdvertisePort)

	conf.Tags["id"] = id
	conf.Tags["role"] = "kappa-server"
	conf.Tags["cluster"] = s.config.ClusterName
	conf.Tags["build"] = s.config.Build
	conf.Tags["port"] = fmt.Sprintf("%d", port)
	if s.config.Bootstrap {
		conf.Tags["bootstrap"] = "1"
	}
	if s.config.BootstrapExpect != 0 {
		conf.Tags["expect"] = fmt.Sprintf("%d", s.config.BootstrapExpect)
	}

	conf.MemberlistConfig.LogOutput = s.config.LogOutput
	conf.LogOutput = s.config.LogOutput
	conf.EventCh = s.serfEventCh
	conf.SnapshotPath = filepath.Join(s.config.DataPath, serfSnapshot)
	conf.ProtocolVersion = conf.ProtocolVersion
	conf.RejoinAfterLeave = true
	conf.EnableNameConflictResolution = false

	conf.Merge = &mergeDelegate{name: s.config.ClusterName}
	if err := ensurePath(conf.SnapshotPath, false); err != nil {
		return nil, err
	}
	return serf.Create(conf)
}

func testAgent(logOutput io.Writer) *Agent {
	if logOutput == nil {
		logOutput = os.Stderr
	}
	config := serf.DefaultConfig()
	config.MemberlistConfig.ProbeInterval = 100 * time.Millisecond
	config.MemberlistConfig.BindAddr = testutil.GetBindAddr().String()
	config.NodeName = config.MemberlistConfig.BindAddr

	agent, err := Create(config, logOutput)
	if err != nil {
		panic(err)
	}
	return agent
}

func testAgent(t *testing.T) *agent.Agent {
	config := serf.DefaultConfig()
	config.MemberlistConfig.BindAddr = testutil.GetBindAddr().String()
	config.NodeName = config.MemberlistConfig.BindAddr

	agent := &agent.Agent{
		RPCAddr:    getRPCAddr(),
		SerfConfig: config,
	}

	if err := agent.Start(); err != nil {
		t.Fatalf("err: %s", err)
	}

	return agent
}