ML-18458: Avoid getting ingestion loops into bad state (#559)

* ML-18458: Avoid getting ingestion loops into bad state

When system is in read-only mode, causing writes to fail, the ingestion loops would get into a bad state, and never recover.

* lint

* More verbose fmt check to find out what's failing lint in CI.

* Fix go version

Author: gtopper

Makefile

@@ -71,11 +71,7 @@ bin:
 
 PHONY: gofmt
 gofmt:
-ifeq ($(shell gofmt -l .),)
-	# gofmt OK
-else
-	$(error Please run `go fmt ./...` to format the code)
-endif
+	if [ "$(gofmt -l .)" != "" ]; then echo 'Please run `go fmt ./...` to format the code'; fi
 
 .PHONY: impi
 impi:

pkg/appender/appender.go

@@ -115,9 +115,8 @@ type MetricsCache struct {
 	asyncAppendChan chan *asyncAppend
 	updatesInFlight int
 
-	metricQueue     *ElasticQueue
-	updatesComplete chan int
-	newUpdates      chan int
+	metricQueue *ElasticQueue
+	newUpdates  chan int
 
 	outstandingUpdates int64
 	requestsInFlight   int64

pkg/appender/ingest.go

@@ -227,19 +227,22 @@ func (mc *MetricsCache) postMetricUpdates(metric *MetricState) {
 	metric.Lock()
 	defer metric.Unlock()
 	var sent bool
+	var err error
 
 	// In case we are in pre get state or our data spreads across multiple partitions, get the new state for the current partition
 	if metric.getState() == storeStatePreGet ||
 		(metric.canSendRequests() && metric.shouldGetState) {
-		sent = mc.sendGetMetricState(metric)
-		if sent {
+		sent, err = mc.sendGetMetricState(metric)
+		if err != nil {
+			metric.setState(storeStateInit)
+		} else if sent {
 			mc.updatesInFlight++
 		}
 	} else if metric.canSendRequests() {
-		sent = mc.writeChunksAndGetState(metric)
+		sent, err = mc.writeChunksAndGetState(metric)
 
 		if !sent {
-			if metric.store.samplesQueueLength() == 0 {
+			if err != nil || metric.store.samplesQueueLength() == 0 {
 				metric.setState(storeStateReady)
 				if metric.store.numNotProcessed == 0 {
 					mc.cacheMetricMap.ResetMetric(metric.hash)
@@ -255,10 +258,10 @@ func (mc *MetricsCache) postMetricUpdates(metric *MetricState) {
 
 }
 
-func (mc *MetricsCache) sendGetMetricState(metric *MetricState) bool {
+func (mc *MetricsCache) sendGetMetricState(metric *MetricState) (bool, error) {
 	// If we are already in a get state, discard
 	if metric.getState() == storeStateGet {
-		return false
+		return false, nil
 	}
 
 	sent, err := metric.store.getChunksState(mc, metric)
@@ -268,14 +271,14 @@ func (mc *MetricsCache) sendGetMetricState(metric *MetricState) bool {
 
 		mc.logger.ErrorWith("Failed to get item state", "metric", metric.Lset, "err", err)
 		setError(mc, metric, err)
-	} else {
-		metric.setState(storeStateGet)
+		return false, err
 	}
+	metric.setState(storeStateGet)
 
-	return sent
+	return sent, nil
 }
 
-func (mc *MetricsCache) writeChunksAndGetState(metric *MetricState) bool {
+func (mc *MetricsCache) writeChunksAndGetState(metric *MetricState) (bool, error) {
 	sent, err := metric.store.writeChunks(mc, metric)
 	if err != nil {
 		// Count errors
@@ -287,13 +290,13 @@ func (mc *MetricsCache) writeChunksAndGetState(metric *MetricState) bool {
 		metric.setState(storeStateUpdate)
 	} else if metric.shouldGetState {
 		// In case we didn't write any data and the metric state needs to be updated, update it straight away
-		sent = mc.sendGetMetricState(metric)
+		sent, err = mc.sendGetMetricState(metric)
 	}
 
 	if sent {
 		mc.updatesInFlight++
 	}
-	return sent
+	return sent, err
 }
 
 // Handle DB responses
@@ -315,6 +318,14 @@ func (mc *MetricsCache) handleResponse(metric *MetricState, resp *v3io.Response,
 			reflect.TypeOf(reqInput), "request", reqInput)
 	}
 
+	clear := func() {
+		resp.Release()
+		metric.store = newChunkStore(mc.logger, metric.Lset.LabelNames(), metric.store.isAggr())
+		metric.retryCount = 0
+		metric.setState(storeStateInit)
+		mc.cacheMetricMap.ResetMetric(metric.hash)
+	}
+
 	if metric.getState() == storeStateGet {
 		// Handle Get response, sync metric state with the DB
 		metric.store.processGetResp(mc, metric, resp)
@@ -328,14 +339,6 @@ func (mc *MetricsCache) handleResponse(metric *MetricState, resp *v3io.Response,
 			}
 			metric.retryCount = 0
 		} else {
-			clear := func() {
-				resp.Release()
-				metric.store = newChunkStore(mc.logger, metric.Lset.LabelNames(), metric.store.isAggr())
-				metric.retryCount = 0
-				metric.setState(storeStateInit)
-				mc.cacheMetricMap.ResetMetric(metric.hash)
-			}
-
 			// Count errors
 			mc.performanceReporter.IncrementCounter("ChunkUpdateRetries", 1)
 
@@ -370,15 +373,24 @@ func (mc *MetricsCache) handleResponse(metric *MetricState, resp *v3io.Response,
 	metric.setState(storeStateReady)
 
 	var sent bool
+	var err error
 
 	// In case our data spreads across multiple partitions, get the new state for the current partition
 	if metric.shouldGetState {
-		sent = mc.sendGetMetricState(metric)
+		sent, err = mc.sendGetMetricState(metric)
+		if err != nil {
+			clear()
+			return false
+		}
 		if sent {
 			mc.updatesInFlight++
 		}
 	} else if canWrite {
-		sent = mc.writeChunksAndGetState(metric)
+		sent, err = mc.writeChunksAndGetState(metric)
+		if err != nil {
+			clear()
+			return false
+		}
 	} else if metric.store.samplesQueueLength() > 0 {
 		mc.metricQueue.Push(metric)
 		metric.setState(storeStateAboutToUpdate)

pkg/appender/store.go

@@ -302,7 +302,8 @@ func (cs *chunkStore) writeChunks(mc *MetricsCache, metric *MetricState) (hasPen
 
 		// Init the partition info and find whether we need to init the metric headers (labels, ..) in the case of a new partition
 		t0 := cs.pending[0].t
-		partition, err := mc.partitionMngr.TimeToPart(t0)
+		var partition *partmgr.DBPartition
+		partition, err = mc.partitionMngr.TimeToPart(t0)
 		if err != nil {
 			hasPendingUpdates = false
 			return

pkg/partmgr/partmgr.go

@@ -233,10 +233,6 @@ func (p *PartitionManager) ReadAndUpdateSchema() (err error) {
 	}
 
 	schemaFilePath := p.GetSchemaFilePath()
-	if err != nil {
-		err = errors.Wrap(err, "Failed to create timer ReadAndUpdateSchemaTimer.")
-		return
-	}
 	schemaInfoResp, err := p.container.GetItemSync(&v3io.GetItemInput{Path: schemaFilePath, AttributeNames: []string{"__mtime_secs", "__mtime_nsecs"}})
 	if err != nil {
 		err = errors.Wrapf(err, "Failed to read schema at path '%s'.", schemaFilePath)