<commit_before>package coordinator

import (
	"common"
	. "common"
	"configuration"
	"datastore"
	"encoding/json"
	"fmt"
	"io/ioutil"
	. "launchpad.net/gocheck"
	"net"
	"net/http"
	"os"
	"protocol"
	"runtime"
	"sort"
	"strconv"
	"strings"
	"testing"
	"time"
)

// Hook up gocheck into the gotest runner.
func Test(t *testing.T) {
	TestingT(t)
}

type CoordinatorSuite struct{}

var _ = Suite(&CoordinatorSuite{})

func init() {
	runtime.GOMAXPROCS(runtime.NumCPU() * 2)
}

const (
	SERVER_STARTUP_TIME = time.Second * 2
	REPLICATION_LAG     = time.Millisecond * 500
)

type DatastoreMock struct {
	datastore.Datastore
	Series          *protocol.Series
	DroppedDatabase string
}

func (self *DatastoreMock) WriteSeriesData(database string, series *protocol.Series) error {
	self.Series = series
	return nil
}

func (self *DatastoreMock) DropDatabase(database string) error {
	self.DroppedDatabase = database
	return nil
}

func (self *DatastoreMock) LogRequestAndAssignSequenceNumber(request *protocol.Request, replicationFactor *uint8, ownerServerId *uint32) error {
	id := uint64(1)
	request.SequenceNumber = &id
	return nil
}

func (self *DatastoreMock) AtomicIncrement(name string, val int) (uint64, error) {
	return uint64(val), nil
}

func (self *DatastoreMock) ReplayRequestsFromSequenceNumber(*uint32, *uint32, *uint32, *uint8, *uint64, func(*[]byte) error) error {
	return nil
}

func stringToSeries(seriesString string, c *C) *protocol.Series {
	series := &protocol.Series{}
	err := json.Unmarshal([]byte(seriesString), &series)
	c.Assert(err, IsNil)
	return series
}

func startAndVerifyCluster(count int, c *C) []*RaftServer {
	firstPort := 0
	servers := make([]*RaftServer, count, count)
	for i := 0; i < count; i++ {
		l, err := net.Listen("tcp4", ":0")
		c.Assert(err, IsNil)
		servers[i] = newConfigAndServer(c)

		if firstPort == 0 {
			firstPort = l.Addr().(*net.TCPAddr).Port
		} else {
			servers[i].config.SeedServers = []string{fmt.Sprintf("http://localhost:%d", firstPort)}
		}

		servers[i].Serve(l)

		time.Sleep(SERVER_STARTUP_TIME)
		// verify that the server is up
		getConfig(servers[i].port, c)
	}
	return servers
}

func cleanWithoutDeleting(servers ...*RaftServer) {
	for _, server := range servers {
		server.Close()
	}
	http.DefaultTransport.(*http.Transport).CloseIdleConnections()
}

func clean(servers ...*RaftServer) {
	cleanWithoutDeleting(servers...)
	for _, server := range servers {
		os.RemoveAll(server.path)
	}
}

func newConfigAndServer(c *C) *RaftServer {
	path, err := ioutil.TempDir(os.TempDir(), "influxdb")
	c.Assert(err, IsNil)
	config := NewClusterConfiguration(&configuration.Configuration{})
	setupConfig := &configuration.Configuration{Hostname: "localhost", RaftDir: path, RaftServerPort: 0}
	server := NewRaftServer(setupConfig, config)
	return server
}

func getConfig(port int, c *C) string {
	host := fmt.Sprintf("localhost:%d", port)
	resp, err := http.Get("http://" + host + "/cluster_config")
	c.Assert(err, Equals, nil)
	defer resp.Body.Close()
	body, err := ioutil.ReadAll(resp.Body)
	c.Assert(err, Equals, nil)
	return string(body)
}

func assertConfigContains(port int, contains string, isContained bool, c *C) {
	body := getConfig(port, c)
	c.Assert(strings.Contains(string(body), contains), Equals, isContained)
}

func (self *CoordinatorSuite) TestCanCreateCoordinatorWithNoSeed(c *C) {
	server := startAndVerifyCluster(1, c)[0]
	defer clean(server)
}

func (self *CoordinatorSuite) TestCanRecover(c *C) {
	server := startAndVerifyCluster(1, c)[0]
	defer clean(server)

	path, port := server.path, server.port

	server.CreateDatabase("db1", uint8(1))
	assertConfigContains(server.port, "db1", true, c)
	cleanWithoutDeleting(server)
	time.Sleep(SERVER_STARTUP_TIME)
	server = newConfigAndServer(c)
	// reset the path and port to the previous server and remove the
	// path that was created by newConfigAndServer
	os.RemoveAll(server.path)
	server.path = path
	server.port = port
	defer clean(server)
	server.ListenAndServe()
	time.Sleep(time.Second)
	assertConfigContains(server.port, "db1", true, c)
}

func (self *CoordinatorSuite) TestCanSnapshot(c *C) {
	server := startAndVerifyCluster(1, c)[0]
	// defer clean(server)

	path, port := server.path, server.port

	for i := 0; i < 1000; i++ {
		dbname := fmt.Sprintf("db%d", i)
		server.CreateDatabase(dbname, uint8(1))
		assertConfigContains(server.port, dbname, true, c)
	}
	size, err := common.GetFileSize(server.raftServer.LogPath())
	c.Assert(err, IsNil)
	server.forceLogCompaction()
	newSize, err := common.GetFileSize(server.raftServer.LogPath())
	c.Assert(err, IsNil)
	c.Assert(newSize < size, Equals, true)
	fmt.Printf("size of %s shrinked from %d to %d\n", server.raftServer.LogPath(), size, newSize)
	cleanWithoutDeleting(server)
	time.Sleep(SERVER_STARTUP_TIME)
	server = newConfigAndServer(c)
	// reset the path and port to the previous server and remove the
	// path that was created by newConfigAndServer
	os.RemoveAll(server.path)
	server.path = path
	server.port = port
	// defer clean(server)
	err = server.ListenAndServe()
	c.Assert(err, IsNil)
	for i := 0; i < 1000; i++ {
		dbname := fmt.Sprintf("db%d", i)
		assertConfigContains(server.port, dbname, true, c)
	}
}
<commit_msg>test snapshot recovery from new peers in the cluster.<commit_after>package coordinator

import (
	. "common"
	"configuration"
	"datastore"
	"encoding/json"
	"fmt"
	"io/ioutil"
	. "launchpad.net/gocheck"
	"net"
	"net/http"
	"os"
	"protocol"
	"runtime"
	"sort"
	"strconv"
	"strings"
	"testing"
	"time"
)

// Hook up gocheck into the gotest runner.
func Test(t *testing.T) {
	TestingT(t)
}

type CoordinatorSuite struct{}

var _ = Suite(&CoordinatorSuite{})

func init() {
	runtime.GOMAXPROCS(runtime.NumCPU() * 2)
}

const (
	SERVER_STARTUP_TIME = time.Second * 2
	REPLICATION_LAG     = time.Millisecond * 500
)

type DatastoreMock struct {
	datastore.Datastore
	Series          *protocol.Series
	DroppedDatabase string
}

func (self *DatastoreMock) WriteSeriesData(database string, series *protocol.Series) error {
	self.Series = series
	return nil
}

func (self *DatastoreMock) DropDatabase(database string) error {
	self.DroppedDatabase = database
	return nil
}

func (self *DatastoreMock) LogRequestAndAssignSequenceNumber(request *protocol.Request, replicationFactor *uint8, ownerServerId *uint32) error {
	id := uint64(1)
	request.SequenceNumber = &id
	return nil
}

func (self *DatastoreMock) AtomicIncrement(name string, val int) (uint64, error) {
	return uint64(val), nil
}

func (self *DatastoreMock) ReplayRequestsFromSequenceNumber(*uint32, *uint32, *uint32, *uint8, *uint64, func(*[]byte) error) error {
	return nil
}

func stringToSeries(seriesString string, c *C) *protocol.Series {
	series := &protocol.Series{}
	err := json.Unmarshal([]byte(seriesString), &series)
	c.Assert(err, IsNil)
	return series
}

func startAndVerifyCluster(count int, c *C) []*RaftServer {
	firstPort := 0
	servers := make([]*RaftServer, count, count)
	for i := 0; i < count; i++ {
		l, err := net.Listen("tcp4", ":0")
		c.Assert(err, IsNil)
		servers[i] = newConfigAndServer(c)

		if firstPort == 0 {
			firstPort = l.Addr().(*net.TCPAddr).Port
		} else {
			servers[i].config.SeedServers = []string{fmt.Sprintf("http://localhost:%d", firstPort)}
		}

		servers[i].Serve(l)

		time.Sleep(SERVER_STARTUP_TIME)
		// verify that the server is up
		getConfig(servers[i].port, c)
	}
	return servers
}

func cleanWithoutDeleting(servers ...*RaftServer) {
	for _, server := range servers {
		server.Close()
	}
	http.DefaultTransport.(*http.Transport).CloseIdleConnections()
}

func clean(servers ...*RaftServer) {
	cleanWithoutDeleting(servers...)
	for _, server := range servers {
		os.RemoveAll(server.path)
	}
}

func newConfigAndServer(c *C) *RaftServer {
	path, err := ioutil.TempDir(os.TempDir(), "influxdb")
	c.Assert(err, IsNil)
	config := NewClusterConfiguration(&configuration.Configuration{})
	setupConfig := &configuration.Configuration{Hostname: "localhost", RaftDir: path, RaftServerPort: 0}
	server := NewRaftServer(setupConfig, config)
	return server
}

func getConfig(port int, c *C) string {
	host := fmt.Sprintf("localhost:%d", port)
	resp, err := http.Get("http://" + host + "/cluster_config")
	c.Assert(err, Equals, nil)
	defer resp.Body.Close()
	body, err := ioutil.ReadAll(resp.Body)
	c.Assert(err, Equals, nil)
	return string(body)
}

func assertConfigContains(port int, contains string, isContained bool, c *C) {
	body := getConfig(port, c)
	c.Assert(strings.Contains(string(body), contains), Equals, isContained)
}

func (self *CoordinatorSuite) TestCanCreateCoordinatorWithNoSeed(c *C) {
	server := startAndVerifyCluster(1, c)[0]
	defer clean(server)
}

func (self *CoordinatorSuite) TestCanRecover(c *C) {
	server := startAndVerifyCluster(1, c)[0]
	defer clean(server)

	path, port := server.path, server.port

	server.CreateDatabase("db1", uint8(1))
	assertConfigContains(server.port, "db1", true, c)
	cleanWithoutDeleting(server)
	time.Sleep(SERVER_STARTUP_TIME)
	server = newConfigAndServer(c)
	// reset the path and port to the previous server and remove the
	// path that was created by newConfigAndServer
	os.RemoveAll(server.path)
	server.path = path
	server.port = port
	defer clean(server)
	server.ListenAndServe()
	time.Sleep(time.Second)
	assertConfigContains(server.port, "db1", true, c)
}

func (self *CoordinatorSuite) TestCanSnapshot(c *C) {
	server := startAndVerifyCluster(1, c)[0]
	// defer clean(server)

	path, port, name := server.path, server.port, server.name

	for i := 0; i < 1000; i++ {
		dbname := fmt.Sprintf("db%d", i)
		server.CreateDatabase(dbname, uint8(1))
		assertConfigContains(server.port, dbname, true, c)
	}
	size, err := GetFileSize(server.raftServer.LogPath())
	c.Assert(err, IsNil)
	server.forceLogCompaction()
	newSize, err := GetFileSize(server.raftServer.LogPath())
	c.Assert(err, IsNil)
	c.Assert(newSize < size, Equals, true)
	fmt.Printf("size of %s shrinked from %d to %d\n", server.raftServer.LogPath(), size, newSize)
	cleanWithoutDeleting(server)
	time.Sleep(SERVER_STARTUP_TIME)
	server = newConfigAndServer(c)
	// reset the path and port to the previous server and remove the
	// path that was created by newConfigAndServer
	os.RemoveAll(server.path)
	server.path = path
	server.port = port
	server.name = name
	// defer clean(server)
	err = server.ListenAndServe()
	c.Assert(err, IsNil)
	time.Sleep(SERVER_STARTUP_TIME)
	for i := 0; i < 1000; i++ {
		dbname := fmt.Sprintf("db%d", i)
		assertConfigContains(server.port, dbname, true, c)
	}

	// make another server join the cluster
	server2 := newConfigAndServer(c)
	defer clean(server2)
	l, err := net.Listen("tcp4", ":0")
	c.Assert(err, IsNil)
	server2.config.SeedServers = []string{fmt.Sprintf("http://localhost:%d", server.port)}
	server2.Serve(l)
	time.Sleep(SERVER_STARTUP_TIME)
	for i := 0; i < 1000; i++ {
		dbname := fmt.Sprintf("db%d", i)
		assertConfigContains(server2.port, dbname, true, c)
	}
}

{'fraction_non_alphanumeric': 0.1405593127399013, 'fraction_numerical': 0.010784134527508682, 'mean_word_length': 4.9699945444626294, 'pattern_counts': {'":': 3, '<': 12, '<?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': '5882334', 'n_tokens_mistral': 4010, 'n_tokens_neox': 3891, 'n_words': 910}