Fix ES bulk processor commit timeout

common/persistence/visibility/store/elasticsearch/processor.go

@@ -199,20 +199,18 @@ func (p *processorImpl) bulkBeforeAction(_ int64, requests []elastic.BulkableReq
  p.metricsHandler.Counter(metrics.ElasticsearchBulkProcessorRequests.GetMetricName()).Record(int64(len(requests)))
  p.metricsHandler.Histogram(metrics.ElasticsearchBulkProcessorBulkSize.GetMetricName(), metrics.ElasticsearchBulkProcessorBulkSize.GetMetricUnit()).
  Record(int64(len(requests)))
- p.metricsHandler.Histogram(metrics.ElasticsearchBulkProcessorQueuedRequests.GetMetricName(), metrics.ElasticsearchBulkProcessorBulkSize.GetMetricUnit()).
- Record(int64(p.mapToAckFuture.Len() - len(requests)))
 
  for _, request := range requests {
  visibilityTaskKey := p.extractVisibilityTaskKey(request)
  if visibilityTaskKey == "" {
  continue
  }
  _, _, _ = p.mapToAckFuture.GetAndDo(visibilityTaskKey, func(key interface{}, value interface{}) error {
- future, ok := value.(*ackFuture)
+ ackF, ok := value.(*ackFuture)
  if !ok {
  p.logger.Fatal(fmt.Sprintf("mapToAckFuture has item of a wrong type %T (%T expected).", value, &ackFuture{}), tag.Value(key))
  }
- future.recordStart(p.metricsHandler)
+ ackF.recordStart(p.metricsHandler)
  return nil
  })
  }
@@ -286,6 +284,10 @@ func (p *processorImpl) bulkAfterAction(_ int64, requests []elastic.BulkableRequ
  p.metricsHandler.Counter(metrics.ElasticsearchBulkProcessorRetries.GetMetricName()).Record(1, metrics.HttpStatusTag(responseItem.Status))
  }
  }
+
+ // Record how many documents are waiting to be flushed to Elasticsearch after this bulk is committed.
+ p.metricsHandler.Histogram(metrics.ElasticsearchBulkProcessorQueuedRequests.GetMetricName(), metrics.ElasticsearchBulkProcessorBulkSize.GetMetricUnit()).
+ Record(int64(p.mapToAckFuture.Len()))
 }
 
 func (p *processorImpl) buildResponseIndex(response *elastic.BulkResponse) map[string]*elastic.BulkResponseItem {
@@ -307,12 +309,12 @@ func (p *processorImpl) buildResponseIndex(response *elastic.BulkResponse) map[s
 func (p *processorImpl) notifyResult(visibilityTaskKey string, ack bool) {
  // Use RemoveIf here to prevent race condition with de-dup logic in Add method.
  _ = p.mapToAckFuture.RemoveIf(visibilityTaskKey, func(key interface{}, value interface{}) bool {
- future, ok := value.(*ackFuture)
+ ackF, ok := value.(*ackFuture)
  if !ok {
  p.logger.Fatal(fmt.Sprintf("mapToAckFuture has item of a wrong type %T (%T expected).", value, &ackFuture{}), tag.ESKey(visibilityTaskKey))
  }
 
- future.done(ack, p.metricsHandler)
+ ackF.done(ack, p.metricsHandler)
  return true
  })
 }

common/persistence/visibility/store/elasticsearch/processor_test.go

@@ -244,6 +244,12 @@ func (s *processorSuite) TestBulkAfterAction_Ack() {
  Items: []map[string]*elastic.BulkResponseItem{mSuccess},
  }
 
+ queuedRequestHistogram := metrics.NewMockHistogramIface(s.controller)
+ s.mockMetricHandler.EXPECT().Histogram(
+ metrics.ElasticsearchBulkProcessorQueuedRequests.GetMetricName(),
+ metrics.ElasticsearchBulkProcessorQueuedRequests.GetMetricUnit(),
+ ).Return(queuedRequestHistogram)
+ queuedRequestHistogram.EXPECT().Record(int64(0))
  s.mockMetricHandler.EXPECT().Timer(metrics.ElasticsearchBulkProcessorRequestLatency.GetMetricName()).Return(metrics.NoopTimerMetricFunc)
  mapVal := newAckFuture()
  s.esProcessor.mapToAckFuture.Put(testKey, mapVal)
@@ -287,6 +293,12 @@ func (s *processorSuite) TestBulkAfterAction_Nack() {
  Items: []map[string]*elastic.BulkResponseItem{mFailed},
  }
 
+ queuedRequestHistogram := metrics.NewMockHistogramIface(s.controller)
+ s.mockMetricHandler.EXPECT().Histogram(
+ metrics.ElasticsearchBulkProcessorQueuedRequests.GetMetricName(),
+ metrics.ElasticsearchBulkProcessorQueuedRequests.GetMetricUnit(),
+ ).Return(queuedRequestHistogram)
+ queuedRequestHistogram.EXPECT().Record(int64(0))
  s.mockMetricHandler.EXPECT().Timer(metrics.ElasticsearchBulkProcessorRequestLatency.GetMetricName()).Return(metrics.NoopTimerMetricFunc)
  mapVal := newAckFuture()
  s.esProcessor.mapToAckFuture.Put(testKey, mapVal)
@@ -352,12 +364,6 @@ func (s *processorSuite) TestBulkBeforeAction() {
  metrics.ElasticsearchBulkProcessorBulkSize.GetMetricUnit(),
  ).Return(bulkSizeHistogram)
  bulkSizeHistogram.EXPECT().Record(int64(1))
- queuedRequestHistorgram := metrics.NewMockHistogramIface(s.controller)
- s.mockMetricHandler.EXPECT().Histogram(
- metrics.ElasticsearchBulkProcessorQueuedRequests.GetMetricName(),
- metrics.ElasticsearchBulkProcessorQueuedRequests.GetMetricUnit(),
- ).Return(queuedRequestHistorgram)
- queuedRequestHistorgram.EXPECT().Record(int64(0))
  s.mockMetricHandler.EXPECT().Timer(metrics.ElasticsearchBulkProcessorWaitAddLatency.GetMetricName()).Return(metrics.NoopTimerMetricFunc)
  s.mockMetricHandler.EXPECT().Timer(metrics.ElasticsearchBulkProcessorWaitStartLatency.GetMetricName()).Return(metrics.NoopTimerMetricFunc)
  mapVal := newAckFuture()
@@ -484,7 +490,7 @@ func (s *processorSuite) TestErrorReasonFromResponse() {
 func (s *processorSuite) Test_End2End() {
  docsCount := 1000
  parallelFactor := 10
- version := int64(2208) //random
+ version := int64(2208) // random
 
  request := &client.BulkableRequest{}
  bulkIndexRequests := make([]elastic.BulkableRequest, docsCount)

common/persistence/visibility/store/elasticsearch/visibility_store.go

@@ -49,11 +49,10 @@ import (
  "go.temporal.io/server/common/persistence/visibility/store/elasticsearch/client"
  "go.temporal.io/server/common/persistence/visibility/store/query"
  "go.temporal.io/server/common/searchattribute"
- "go.temporal.io/server/common/util"
 )
 
 const (
- persistenceName = "elasticsearch"
+ PersistenceName = "elasticsearch"
 
  delimiter = "~"
  pointInTimeKeepAliveInterval = "1m"
@@ -128,7 +127,7 @@ func (s *visibilityStore) Close() {
 }
 
 func (s *visibilityStore) GetName() string {
- return persistenceName
+ return PersistenceName
 }
 
 func (s *visibilityStore) RecordWorkflowExecutionStarted(
@@ -232,32 +231,30 @@ func (s *visibilityStore) addBulkIndexRequestAndWait(
 }
 
 func (s *visibilityStore) addBulkRequestAndWait(
- ctx context.Context,
+ _ context.Context,
  bulkRequest *client.BulkableRequest,
  visibilityTaskKey string,
 ) error {
  s.checkProcessor()
 
  // Add method is blocking. If bulk processor is busy flushing previous bulk, request will wait here.
- // Therefore, ackTimeoutTimer in fact wait for request to be committed after it was added to bulk processor.
- // TODO: this also means ctx is not respected if bulk processor is busy. Shall we make Add non-blocking or
- // respecting the context?
- future := s.processor.Add(bulkRequest, visibilityTaskKey)
-
- ackTimeout := s.processorAckTimeout()
- if deadline, ok := ctx.Deadline(); ok {
- ackTimeout = util.Min(ackTimeout, time.Until(deadline))
- }
- subCtx, subCtxCancelFn := context.WithTimeout(context.Background(), ackTimeout)
- defer subCtxCancelFn()
-
- ack, err := future.Get(subCtx)
+ ackF := s.processor.Add(bulkRequest, visibilityTaskKey)
 
- if errors.Is(err, context.DeadlineExceeded) {
- return &persistence.TimeoutError{Msg: fmt.Sprintf("visibility task %s timedout waiting for ACK after %v", visibilityTaskKey, s.processorAckTimeout())}
- }
+ // processorAckTimeout is a maximum duration for bulk processor to commit the bulk and unblock the `ackF`.
+ // Default value is 30s and this timeout should never have happened,
+ // because Elasticsearch must process a bulk within 30s.
+ // Parent context is not respected here because it has shorter timeout (3s),
+ // which might already expired here due to wait at Add method above.
+ ctx, cancel := context.WithTimeout(context.Background(), s.processorAckTimeout())
+ defer cancel()
+ ack, err := ackF.Get(ctx)
 
  if err != nil {
+ if errors.Is(err, context.DeadlineExceeded) {
+ return &persistence.TimeoutError{Msg: fmt.Sprintf("visibility task %s timed out waiting for ACK after %v", visibilityTaskKey, s.processorAckTimeout())}
+ }
+ // Returns non-retryable Internal error here because these errors are unexpected.
+ // Visibility task processor retries all errors though, therefore new request will be generated for the same task.
  return serviceerror.NewInternal(fmt.Sprintf("visibility task %s received error %v", visibilityTaskKey, err))
  }
 
@@ -900,7 +897,7 @@ func (s *visibilityStore) parseESDoc(docID string, docSource json.RawMessage, sa
  // Very important line. See finishParseJSONValue bellow.
  d.UseNumber()
  if err := d.Decode(&sourceMap); err != nil {
- s.metricsHandler.Counter(metrics.ElasticsearchDocumentParseFailuresCount.GetMetricName()) //.Record(1)
+ s.metricsHandler.Counter(metrics.ElasticsearchDocumentParseFailuresCount.GetMetricName()).Record(1)
  return nil, serviceerror.NewInternal(fmt.Sprintf("Unable to unmarshal JSON from Elasticsearch document(%s): %v", docID, err))
  }
 

service/history/configs/config.go

@@ -498,14 +498,14 @@ func NewConfig(dc *dynamicconfig.Collection, numberOfShards int32, isAdvancedVis
  SearchAttributesTotalSizeLimit: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.SearchAttributesTotalSizeLimit, 40*1024),
  IndexerConcurrency: dc.GetIntProperty(dynamicconfig.WorkerIndexerConcurrency, 100),
  ESProcessorNumOfWorkers: dc.GetIntProperty(dynamicconfig.WorkerESProcessorNumOfWorkers, 1),
- // Should be not greater than NumberOfShards(512)/NumberOfHistoryNodes(4) * VisibilityTaskWorkerCount(10)/ESProcessorNumOfWorkers(1) divided by workflow distribution factor (2 at least).
+ // Should not be greater than number of visibility task queue workers VisibilityProcessorSchedulerWorkerCount (default 512)
  // Otherwise, visibility queue processors won't be able to fill up bulk with documents (even under heavy load) and bulk will flush due to interval, not number of actions.
  ESProcessorBulkActions: dc.GetIntProperty(dynamicconfig.WorkerESProcessorBulkActions, 500),
  // 16MB - just a sanity check. With ES document size ~1Kb it should never be reached.
  ESProcessorBulkSize: dc.GetIntProperty(dynamicconfig.WorkerESProcessorBulkSize, 16*1024*1024),
  // Bulk processor will flush every this interval regardless of last flush due to bulk actions.
  ESProcessorFlushInterval: dc.GetDurationProperty(dynamicconfig.WorkerESProcessorFlushInterval, 1*time.Second),
- ESProcessorAckTimeout: dc.GetDurationProperty(dynamicconfig.WorkerESProcessorAckTimeout, 1*time.Minute),
+ ESProcessorAckTimeout: dc.GetDurationProperty(dynamicconfig.WorkerESProcessorAckTimeout, 30*time.Second),
 
  EnableCrossNamespaceCommands: dc.GetBoolProperty(dynamicconfig.EnableCrossNamespaceCommands, true),
  EnableActivityEagerExecution: dc.GetBoolPropertyFnWithNamespaceFilter(dynamicconfig.EnableActivityEagerExecution, false),

service/history/visibilityQueueTaskExecutor.go

@@ -319,10 +319,10 @@ func (t *visibilityQueueTaskExecutor) upsertExecution(
 }
 
 func (t *visibilityQueueTaskExecutor) processCloseExecution(
- ctx context.Context,
+ parentCtx context.Context,
  task *tasks.CloseExecutionVisibilityTask,
 ) (retError error) {
- ctx, cancel := context.WithTimeout(ctx, taskTimeout)
+ ctx, cancel := context.WithTimeout(parentCtx, taskTimeout)
  defer cancel()
 
  namespaceEntry, err := t.shard.GetNamespaceRegistry().GetNamespaceByID(namespace.ID(task.GetNamespaceID()))
@@ -396,8 +396,13 @@ func (t *visibilityQueueTaskExecutor) processCloseExecution(
  return err
  }
 
+ // Elasticsearch bulk processor doesn't respect context timeout
+ // because under heavy load bulk flush might take longer than taskTimeout.
+ // Therefore, ctx timeout might be already expired
+ // and parentCtx (which doesn't have timeout) must be used everywhere bellow.
+
  if t.enableCloseWorkflowCleanup(namespaceEntry.Name().String()) {
- return t.cleanupExecutionInfo(ctx, task)
+ return t.cleanupExecutionInfo(parentCtx, task)
  }
  return nil
 }
@@ -500,6 +505,9 @@ func (t *visibilityQueueTaskExecutor) cleanupExecutionInfo(
  ctx context.Context,
  task *tasks.CloseExecutionVisibilityTask,
 ) (retError error) {
+ ctx, cancel := context.WithTimeout(ctx, taskTimeout)
+ defer cancel()
+
  weContext, release, err := getWorkflowExecutionContextForTask(ctx, t.cache, task)
  if err != nil {
  return err

service/worker/deletenamespace/reclaimresources/activities.go

@@ -37,6 +37,7 @@ import (
  "go.temporal.io/server/common/namespace"
  "go.temporal.io/server/common/persistence"
  "go.temporal.io/server/common/persistence/visibility/manager"
+ "go.temporal.io/server/common/persistence/visibility/store/elasticsearch"
  "go.temporal.io/server/service/worker/deletenamespace/errors"
 )
 
@@ -63,7 +64,7 @@ func NewActivities(
  }
 }
 func (a *Activities) IsAdvancedVisibilityActivity(_ context.Context) (bool, error) {
- return strings.Contains(a.visibilityManager.GetName(), "elasticsearch"), nil
+ return strings.Contains(a.visibilityManager.GetName(), elasticsearch.PersistenceName), nil
 }
 
 func (a *Activities) CountExecutionsAdvVisibilityActivity(ctx context.Context, nsID namespace.ID, nsName namespace.Name) (int64, error) {