Automatically pause queue if issue index service is unavailable

Author: lafriks

modules/indexer/issues/bleve.go

@@ -186,6 +186,15 @@ func (b *BleveIndexer) Init() (bool, error) {
 	return false, err
 }
 
+// SetAvailabilityChangeCallback does nothing
+func (b *BleveIndexer) SetAvailabilityChangeCallback(callback func(bool)) {
+}
+
+// Ping does nothing
+func (b *BleveIndexer) Ping() bool {
+	return true
+}
+
 // Close will close the bleve indexer
 func (b *BleveIndexer) Close() {
 	if b.indexer != nil {

modules/indexer/issues/db.go

@@ -14,6 +14,15 @@ func (db *DBIndexer) Init() (bool, error) {
 	return false, nil
 }
 
+// SetAvailabilityChangeCallback dummy function
+func (db *DBIndexer) SetAvailabilityChangeCallback(callback func(bool)) {
+}
+
+// Ping checks if database is available
+func (db *DBIndexer) Ping() bool {
+	return models.Ping() != nil
+}
+
 // Index dummy function
 func (db *DBIndexer) Index(issue []*IndexerData) error {
 	return nil

modules/indexer/issues/elastic_search.go

@@ -20,8 +20,11 @@ var _ Indexer = &ElasticSearchIndexer{}
 
 // ElasticSearchIndexer implements Indexer interface
 type ElasticSearchIndexer struct {
-	client      *elastic.Client
-	indexerName string
+	client               *elastic.Client
+	indexerName          string
+	available            bool
+	availabilityCallback func(bool)
+	stopTimer            chan struct{}
 }
 
 type elasticLogger struct {
@@ -56,10 +59,27 @@ func NewElasticSearchIndexer(url, indexerName string) (*ElasticSearchIndexer, er
 		return nil, err
 	}
 
-	return &ElasticSearchIndexer{
+	indexer := &ElasticSearchIndexer{
 		client:      client,
 		indexerName: indexerName,
-	}, nil
+		available:   true,
+		stopTimer:   make(chan struct{}),
+	}
+
+	ticker := time.NewTicker(10 * time.Second)
+	go func() {
+		for {
+			select {
+			case <-ticker.C:
+				indexer.checkAvailability()
+			case <-indexer.stopTimer:
+				ticker.Stop()
+				return
+			}
+		}
+	}()
+
+	return indexer, nil
 }
 
 const (
@@ -96,15 +116,15 @@ func (b *ElasticSearchIndexer) Init() (bool, error) {
 	ctx := context.Background()
 	exists, err := b.client.IndexExists(b.indexerName).Do(ctx)
 	if err != nil {
-		return false, err
+		return false, b.checkError(err)
 	}
 
 	if !exists {
 		mapping := defaultMapping
 
 		createIndex, err := b.client.CreateIndex(b.indexerName).BodyString(mapping).Do(ctx)
 		if err != nil {
-			return false, err
+			return false, b.checkError(err)
 		}
 		if !createIndex.Acknowledged {
 			return false, errors.New("init failed")
@@ -115,6 +135,16 @@ func (b *ElasticSearchIndexer) Init() (bool, error) {
 	return true, nil
 }
 
+// SetAvailabilityChangeCallback sets callback that will be triggered when availability changes
+func (b *ElasticSearchIndexer) SetAvailabilityChangeCallback(callback func(bool)) {
+	b.availabilityCallback = callback
+}
+
+// Ping checks if elastic is available
+func (b *ElasticSearchIndexer) Ping() bool {
+	return b.available
+}
+
 // Index will save the index data
 func (b *ElasticSearchIndexer) Index(issues []*IndexerData) error {
 	if len(issues) == 0 {
@@ -132,7 +162,7 @@ func (b *ElasticSearchIndexer) Index(issues []*IndexerData) error {
 				"comments": issue.Comments,
 			}).
 			Do(context.Background())
-		return err
+		return b.checkError(err)
 	}
 
 	reqs := make([]elastic.BulkableRequest, 0)
@@ -155,7 +185,7 @@ func (b *ElasticSearchIndexer) Index(issues []*IndexerData) error {
 		Index(b.indexerName).
 		Add(reqs...).
 		Do(context.Background())
-	return err
+	return b.checkError(err)
 }
 
 // Delete deletes indexes by ids
@@ -167,7 +197,7 @@ func (b *ElasticSearchIndexer) Delete(ids ...int64) error {
 			Index(b.indexerName).
 			Id(fmt.Sprintf("%d", ids[0])).
 			Do(context.Background())
-		return err
+		return b.checkError(err)
 	}
 
 	reqs := make([]elastic.BulkableRequest, 0)
@@ -183,7 +213,7 @@ func (b *ElasticSearchIndexer) Delete(ids ...int64) error {
 		Index(b.indexerName).
 		Add(reqs...).
 		Do(context.Background())
-	return err
+	return b.checkError(err)
 }
 
 // Search searches for issues by given conditions.
@@ -207,7 +237,7 @@ func (b *ElasticSearchIndexer) Search(keyword string, repoIDs []int64, limit, st
 		From(start).Size(limit).
 		Do(context.Background())
 	if err != nil {
-		return nil, err
+		return nil, b.checkError(err)
 	}
 
 	hits := make([]Match, 0, limit)
@@ -225,4 +255,32 @@ func (b *ElasticSearchIndexer) Search(keyword string, repoIDs []int64, limit, st
 }
 
 // Close implements indexer
-func (b *ElasticSearchIndexer) Close() {}
+func (b *ElasticSearchIndexer) Close() {
+	close(b.stopTimer)
+}
+
+func (b *ElasticSearchIndexer) checkError(err error) error {
+	if elastic.IsConnErr(err) && b.available {
+		b.available = false
+		if b.availabilityCallback != nil {
+			b.availabilityCallback(b.available)
+		}
+	}
+	return err
+}
+
+func (b *ElasticSearchIndexer) checkAvailability() {
+	if b.available {
+		return
+	}
+
+	// Request cluster state to check if elastic is available again
+	_, err := b.client.ClusterState().Do(context.Background())
+	if err != nil {
+		return
+	}
+	b.available = true
+	if b.availabilityCallback != nil {
+		b.availabilityCallback(b.available)
+	}
+}

modules/indexer/issues/indexer.go

@@ -47,6 +47,8 @@ type SearchResult struct {
 // Indexer defines an interface to indexer issues contents
 type Indexer interface {
 	Init() (bool, error)
+	Ping() bool
+	SetAvailabilityChangeCallback(callback func(bool))
 	Index(issue []*IndexerData) error
 	Delete(ids ...int64) error
 	Search(kw string, repoIDs []int64, limit, start int) (*SearchResult, error)
@@ -111,6 +113,7 @@ func InitIssueIndexer(syncReindex bool) {
 			}
 
 			iData := make([]*IndexerData, 0, len(data))
+			unhandled := make([]queue.Data, 0, len(data))
 			for _, datum := range data {
 				indexerData, ok := datum.(*IndexerData)
 				if !ok {
@@ -119,13 +122,34 @@ func InitIssueIndexer(syncReindex bool) {
 				}
 				log.Trace("IndexerData Process: %d %v %t", indexerData.ID, indexerData.IDs, indexerData.IsDelete)
 				if indexerData.IsDelete {
-					_ = indexer.Delete(indexerData.IDs...)
+					if err := indexer.Delete(indexerData.IDs...); err != nil {
+						log.Error("Error whilst deleting from index: %v Error: %v", indexerData.IDs, err)
+						if indexer.Ping() {
+							continue
+						}
+						// Add back to queue
+						unhandled = append(unhandled, datum)
+					}
 					continue
 				}
 				iData = append(iData, indexerData)
 			}
+			if len(unhandled) > 0 {
+				for _, indexerData := range iData {
+					unhandled = append(unhandled, indexerData)
+				}
+				return unhandled
+			}
 			if err := indexer.Index(iData); err != nil {
 				log.Error("Error whilst indexing: %v Error: %v", iData, err)
+				if indexer.Ping() {
+					return nil
+				}
+				// Add back to queue
+				for _, indexerData := range iData {
+					unhandled = append(unhandled, indexerData)
+				}
+				return unhandled
 			}
 			return nil
 		}
@@ -193,6 +217,18 @@ func InitIssueIndexer(syncReindex bool) {
 			log.Fatal("Unknown issue indexer type: %s", setting.Indexer.IssueType)
 		}
 
+		if queue, ok := issueIndexerQueue.(queue.Pausable); ok {
+			holder.get().SetAvailabilityChangeCallback(func(available bool) {
+				if !available {
+					log.Info("Issue index queue paused")
+					queue.Pause()
+				} else {
+					log.Info("Issue index queue resumed")
+					queue.Resume()
+				}
+			})
+		}
+
 		// Start processing the queue
 		go graceful.GetManager().RunWithShutdownFns(issueIndexerQueue.Run)