es: stop client is rolling test and add debug log for rolling index

Author: safchain

storage/elasticsearch/client.go

@@ -82,8 +82,6 @@ type Client struct {
 	esClient      *elastic.Client
 	bulkProcessor *elastic.BulkProcessor
 	started       atomic.Value
-	quit          chan bool
-	wg            sync.WaitGroup
 	cfg           Config
 	indices       map[string]Index
 	rollService   *rollIndexService
@@ -405,17 +403,12 @@ func (c *Client) Start() {
 // Stop Elasticsearch background client
 func (c *Client) Stop() {
 	if c.started.Load() == true {
-		c.quit <- true
-		c.wg.Wait()
+		if c.rollService != nil {
+			c.rollService.stop()
+		}
 
 		c.esClient.Stop()
 	}
-
-	c.RLock()
-	for _, l := range c.listeners {
-		l.OnStarted()
-	}
-	c.RUnlock()
 }
 
 // Started is the client already started ?
@@ -467,7 +460,6 @@ func NewClient(indices []Index, cfg Config, electionService common.MasterElectio
 
 	client := &Client{
 		url:     url,
-		quit:    make(chan bool, 1),
 		cfg:     cfg,
 		indices: indicesMap,
 	}

storage/elasticsearch/rollindex.go

@@ -80,7 +80,7 @@ func (r *rollIndexService) cleanup(index Index) {
 }
 
 func (r *rollIndexService) roll(force bool) {
-	logging.GetLogger().Debug("Start rolling indices...")
+	logging.GetLogger().Debugf("Start rolling indices (forced: %v)...", force)
 
 	for _, index := range r.indices {
 		ri := r.client.esClient.RolloverIndex(index.Alias())
@@ -104,6 +104,8 @@ func (r *rollIndexService) roll(force bool) {
 		if needToRoll {
 			logging.GetLogger().Infof("Index %s rolling over", index.Alias())
 
+			logging.GetLogger().Debugf("Rolling over with: %+v", ri)
+
 			resp, err := ri.Do(context.Background())
 			if err != nil {
 				logging.GetLogger().Errorf("Error while rolling index %s: %s", index.Alias(), err)

tests/elasticsearch_test.go

@@ -101,6 +101,7 @@ func TestRollingSimple(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer client.Stop()
 
 	for i := 0; i != 9; i++ {
 		id := fmt.Sprintf("before-entry_%d", i)