cleanup

Author: timl3136

internal/internal_event_handlers.go

@@ -943,7 +943,6 @@ func (weh *workflowExecutionEventHandlerImpl) ProcessEvent(
 
 	historySum := weh.estimateHistorySize(event)
 	weh.workflowInfo.TotalHistoryBytes += int64(historySum)
-	weh.logger.Info("EstimateHistorySize", zap.Int("historyBytes", historySum), zap.Int64("totalHistoryBytes", weh.workflowInfo.TotalHistoryBytes))
 
 	// When replaying histories to get stack trace or current state the last event might be not
 	// decision started. So always call OnDecisionTaskStarted on the last event.

internal/internal_task_handlers.go

@@ -51,8 +51,6 @@ const (
 	defaultStickyCacheSize = 10000
 
 	noRetryBackoff = time.Duration(-1)
-
-	historySizeEstimationOffset = 1 * 100
 )
 
 type (
@@ -271,7 +269,6 @@ func (eh *history) NextDecisionEvents() (result []*s.HistoryEvent, markers []*s.
 	if len(result) > 0 {
 		eh.next, markers, err = eh.nextDecisionEvents()
 	}
-
 	return result, markers, checksum, err
 }
 
@@ -832,7 +829,6 @@ func (w *workflowExecutionContextImpl) ProcessWorkflowTask(workflowTask *workflo
 	if err := w.ResetIfStale(task, historyIterator); err != nil {
 		return nil, err
 	}
-
 	w.SetCurrentTask(task)
 
 	eventHandler := w.getEventHandler()
@@ -874,7 +870,6 @@ ProcessEvents:
 				if err != nil {
 					return nil, err
 				}
-
 				if w.isWorkflowCompleted {
 					break ProcessEvents
 				}

internal/internal_task_pollers.go

@@ -110,16 +110,15 @@ type (
 	}
 
 	historyIteratorImpl struct {
-		iteratorFunc      func(nextPageToken []byte) (*s.History, []byte, error)
-		execution         *s.WorkflowExecution
-		nextPageToken     []byte
-		domain            string
-		service           workflowserviceclient.Interface
-		metricsScope      tally.Scope
-		startedEventID    int64
-		maxEventID        int64 // Equivalent to History Count
-		featureFlags      FeatureFlags
-		totalHistoryBytes int64
+		iteratorFunc   func(nextPageToken []byte) (*s.History, []byte, error)
+		execution      *s.WorkflowExecution
+		nextPageToken  []byte
+		domain         string
+		service        workflowserviceclient.Interface
+		metricsScope   tally.Scope
+		startedEventID int64
+		maxEventID     int64 // Equivalent to History Count
+		featureFlags   FeatureFlags
 	}
 
 	localActivityTaskPoller struct {
@@ -331,7 +330,6 @@ func (wtp *workflowTaskPoller) processWorkflowTask(task *workflowTask) error {
 		})
 		return nil
 	}
-	// TODO: get workflowinfo
 	doneCh := make(chan struct{})
 	laResultCh := make(chan *localActivityResult)
 	// close doneCh so local activity worker won't get blocked forever when trying to send back result to laResultCh.
@@ -856,15 +854,14 @@ func (wtp *workflowTaskPoller) toWorkflowTask(response *s.PollForDecisionTaskRes
 		}
 	}
 	historyIterator := &historyIteratorImpl{
-		nextPageToken:     response.NextPageToken,
-		execution:         response.WorkflowExecution,
-		domain:            wtp.domain,
-		service:           wtp.service,
-		metricsScope:      wtp.metricsScope,
-		startedEventID:    startEventID,
-		maxEventID:        nextEventID - 1,
-		featureFlags:      wtp.featureFlags,
-		totalHistoryBytes: response.GetTotalHistoryBytes(),
+		nextPageToken:  response.NextPageToken,
+		execution:      response.WorkflowExecution,
+		domain:         wtp.domain,
+		service:        wtp.service,
+		metricsScope:   wtp.metricsScope,
+		startedEventID: startEventID,
+		maxEventID:     nextEventID - 1,
+		featureFlags:   wtp.featureFlags,
 	}
 	task := &workflowTask{
 		task:            response,
@@ -902,11 +899,6 @@ func (h *historyIteratorImpl) HasNextPage() bool {
 	return h.nextPageToken != nil
 }
 
-// GetTotalHistoryBytes returns History Size of current history
-func (h *historyIteratorImpl) GetTotalHistoryBytes() int64 {
-	return h.totalHistoryBytes
-}
-
 // GetHistoryCount returns History Event Count of current history (aka maxEventID)
 func (h *historyIteratorImpl) GetHistoryCount() int64 {
 	return h.maxEventID

internal/workflow.go

@@ -1161,13 +1161,13 @@ func (wc *workflowEnvironmentInterceptor) GetMetricsScope(ctx Context) tally.Sco
 	return wc.env.GetMetricsScope()
 }
 
-// GetTotalHistoryBytes returns the current history size of that workflow
-func GetTotalHistoryBytes(ctx Context) int64 {
+// GetTotalEstimatedHistoryBytes returns the current history size of that workflow
+func GetTotalEstimatedHistoryBytes(ctx Context) int64 {
 	i := getWorkflowInterceptor(ctx)
 	return i.GetWorkflowInfo(ctx).TotalHistoryBytes
 }
 
-// GetHistoryCount returns the current number of events of that workflow
+// GetHistoryCount returns the current number of history events of that workflow
 func GetHistoryCount(ctx Context) int64 {
 	i := getWorkflowInterceptor(ctx)
 	return i.GetWorkflowInfo(ctx).HistoryCount

workflow/workflow.go

@@ -219,7 +219,7 @@ func GetMetricsScope(ctx Context) tally.Scope {
 
 // GetTotalHistoryBytes returns the current history size of that workflow
 func GetTotalHistoryBytes(ctx Context) int64 {
-	return internal.GetTotalHistoryBytes(ctx)
+	return internal.GetTotalEstimatedHistoryBytes(ctx)
 }
 
 // GetHistoryCount returns the current number of history event of that workflow