<commit_before>package cluster

import (
	"context"
	"crypto/tls"
	"fmt"
	"sync"
	"time"

	"github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/embed"
	"github.com/coreos/etcd/etcdserver/api/v3client"
	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
	"github.com/coreos/etcd/pkg/types"
	"github.com/coreos/etcdlabs/clusterpb"

	humanize "github.com/dustin/go-humanize"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

// Member contains *embed.Etcd and its state.
type Member struct {
	clus *Cluster
	cfg  *embed.Config
	srv  *embed.Etcd

	stoppedStartedAt time.Time

	statusLock sync.RWMutex
	status     clusterpb.MemberStatus
}

// Start starts the member.
func (m *Member) Start() error {
	srv, err := embed.StartEtcd(m.cfg)
	if err != nil {
		return err
	}
	m.srv = srv

	// copy and overwrite with internal configuration
	// in case it was configured with auto TLS
	nc := m.srv.Config()
	m.cfg = &nc

	var rerr error
	select {
	case <-m.srv.Server.ReadyNotify():
	case rerr = <-m.srv.Err():
	case <-m.srv.Server.StopNotify():
		rerr = fmt.Errorf("received from etcdserver.Server.StopNotify")
	}
	if rerr != nil {
		return rerr
	}

	m.stoppedStartedAt = time.Now()

	m.statusLock.Lock()
	m.status.State = clusterpb.FollowerMemberStatus
	m.status.StateTxt = fmt.Sprintf("%s just started (%s)", m.status.Name, humanize.Time(m.stoppedStartedAt))
	m.status.IsLeader = false
	m.statusLock.Unlock()

	plog.Printf("started %s (client %s, peer %s)", m.cfg.Name, m.cfg.LCUrls[0].String(), m.cfg.LPUrls[0].String())
	return nil
}

// Restart restarts the member.
func (m *Member) Restart() error {
	plog.Printf("restarting %q(%s)", m.cfg.Name, m.srv.Server.ID().String())

	m.statusLock.RLock()
	if m.status.State != clusterpb.StoppedMemberStatus {
		plog.Warningf("%s is already started", m.cfg.Name)
		m.statusLock.RUnlock()
		return nil
	}
	m.statusLock.RUnlock()

	m.cfg.ClusterState = embed.ClusterStateFlagExisting

	// start server
	srv, err := embed.StartEtcd(m.cfg)
	if err != nil {
		return err
	}
	m.srv = srv

	nc := m.srv.Config()
	m.cfg = &nc

	// this blocks when quorum is lost
	// <-m.srv.Server.ReadyNotify()

	m.stoppedStartedAt = time.Now()

	m.statusLock.Lock()
	m.status.IsLeader = false
	m.status.State = clusterpb.FollowerMemberStatus
	m.status.StateTxt = fmt.Sprintf("%s just restarted (%s)", m.status.Name, humanize.Time(m.stoppedStartedAt))
	m.statusLock.Unlock()

	plog.Printf("restarted %q(%s)", m.cfg.Name, m.srv.Server.ID().String())
	return nil
}

// Stop stops the member.
func (m *Member) Stop() {
	plog.Printf("stopping %q(%s)", m.cfg.Name, m.srv.Server.ID().String())

	m.statusLock.RLock()
	if m.status.State == clusterpb.StoppedMemberStatus {
		plog.Warningf("%s is already stopped", m.cfg.Name)
		m.statusLock.RUnlock()
		return
	}
	m.statusLock.RUnlock()

	m.stoppedStartedAt = time.Now()

	m.statusLock.Lock()
	m.status.IsLeader = false
	m.status.State = clusterpb.StoppedMemberStatus
	m.status.StateTxt = fmt.Sprintf("%s just stopped (%s)", m.status.Name, humanize.Time(m.stoppedStartedAt))
	m.status.DBSize = 0
	m.status.DBSizeTxt = ""
	m.status.Hash = 0
	m.statusLock.Unlock()

	// TODO: stop with/without leadership transfer?
	m.srv.Server.HardStop()
	m.srv.Close()

	var cerr error
	select {
	case cerr = <-m.srv.Err():
	case <-m.srv.Server.StopNotify():
		cerr = fmt.Errorf("received from EtcdServer.StopNotify")<commit_msg>cluster: gracefully shutdown gRPC server<commit_after>package cluster

import (
	"context"
	"crypto/tls"
	"fmt"
	"sync"
	"time"

	"github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/embed"
	"github.com/coreos/etcd/etcdserver/api/v3client"
	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
	"github.com/coreos/etcd/pkg/types"
	"github.com/coreos/etcdlabs/clusterpb"

	humanize "github.com/dustin/go-humanize"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

// Member contains *embed.Etcd and its state.
type Member struct {
	clus *Cluster
	cfg  *embed.Config
	srv  *embed.Etcd

	stoppedStartedAt time.Time

	statusLock sync.RWMutex
	status     clusterpb.MemberStatus
}

// Start starts the member.
func (m *Member) Start() error {
	srv, err := embed.StartEtcd(m.cfg)
	if err != nil {
		return err
	}
	m.srv = srv

	// copy and overwrite with internal configuration
	// in case it was configured with auto TLS
	nc := m.srv.Config()
	m.cfg = &nc

	var rerr error
	select {
	case <-m.srv.Server.ReadyNotify():
	case rerr = <-m.srv.Err():
	case <-m.srv.Server.StopNotify():
		rerr = fmt.Errorf("received from etcdserver.Server.StopNotify")
	}
	if rerr != nil {
		return rerr
	}

	m.stoppedStartedAt = time.Now()

	m.statusLock.Lock()
	m.status.State = clusterpb.FollowerMemberStatus
	m.status.StateTxt = fmt.Sprintf("%s just started (%s)", m.status.Name, humanize.Time(m.stoppedStartedAt))
	m.status.IsLeader = false
	m.statusLock.Unlock()

	plog.Printf("started %s (client %s, peer %s)", m.cfg.Name, m.cfg.LCUrls[0].String(), m.cfg.LPUrls[0].String())
	return nil
}

// Restart restarts the member.
func (m *Member) Restart() error {
	plog.Printf("restarting %q(%s)", m.cfg.Name, m.srv.Server.ID().String())

	m.statusLock.RLock()
	if m.status.State != clusterpb.StoppedMemberStatus {
		plog.Warningf("%s is already started", m.cfg.Name)
		m.statusLock.RUnlock()
		return nil
	}
	m.statusLock.RUnlock()

	m.cfg.ClusterState = embed.ClusterStateFlagExisting

	// start server
	srv, err := embed.StartEtcd(m.cfg)
	if err != nil {
		return err
	}
	m.srv = srv

	nc := m.srv.Config()
	m.cfg = &nc

	// this blocks when quorum is lost
	// <-m.srv.Server.ReadyNotify()

	m.stoppedStartedAt = time.Now()

	m.statusLock.Lock()
	m.status.IsLeader = false
	m.status.State = clusterpb.FollowerMemberStatus
	m.status.StateTxt = fmt.Sprintf("%s just restarted (%s)", m.status.Name, humanize.Time(m.stoppedStartedAt))
	m.statusLock.Unlock()

	plog.Printf("restarted %q(%s)", m.cfg.Name, m.srv.Server.ID().String())
	return nil
}

// Stop stops the member.
func (m *Member) Stop() {
	plog.Printf("stopping %q(%s)", m.cfg.Name, m.srv.Server.ID().String())

	m.statusLock.RLock()
	if m.status.State == clusterpb.StoppedMemberStatus {
		plog.Warningf("%s is already stopped", m.cfg.Name)
		m.statusLock.RUnlock()
		return
	}
	m.statusLock.RUnlock()

	m.stoppedStartedAt = time.Now()

	m.statusLock.Lock()
	m.status.IsLeader = false
	m.status.State = clusterpb.StoppedMemberStatus
	m.status.StateTxt = fmt.Sprintf("%s just stopped (%s)", m.status.Name, humanize.Time(m.stoppedStartedAt))
	m.status.DBSize = 0
	m.status.DBSizeTxt = ""
	m.status.Hash = 0
	m.statusLock.Unlock()

	// TODO: stop with/without leadership transfer?
	// m.srv.Server.HardStop()

	// stops embedded server to trigger
	// gRPC server graceful shutdown
	m.srv.Close()

	var cerr error
	select {
	case cerr = <-m.srv.Err():
	case <-m.srv.Server.StopNotify():
		cerr = fmt.Errorf("received from EtcdServer.StopNotify")
{'fraction_non_alphanumeric': 0.15722996515679444, 'fraction_numerical': 0.0017421602787456446, 'mean_word_length': 3.8377808988764044, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2176702', 'n_tokens_mistral': 2775, 'n_tokens_neox': 2676, 'n_words': 530}