update pinot visibility triple manager to write to pinot and ES

common/constants.go

@@ -150,6 +150,8 @@ const (
 	AdvancedVisibilityWritingModeOn = "on"
 	// AdvancedVisibilityWritingModeDual means write to both normal visibility and advanced visibility store
 	AdvancedVisibilityWritingModeDual = "dual"
+	// AdvacnedVisibilityWritingModeTriple means write to both normal visibility and advanced visibility store, includes ES and Pinot
+	AdvacnedVisibilityWritingModeTriple = "triple"
 )
 
 const (

common/persistence/pinotVisibilityTripleManager.go

@@ -38,6 +38,7 @@ type (
 		pinotVisibilityManager VisibilityManager
 		esVisibilityManager    VisibilityManager
 		readModeIsFromPinot    dynamicconfig.BoolPropertyFnWithDomainFilter
+		readModeIsFromES       dynamicconfig.BoolPropertyFnWithDomainFilter
 		writeMode              dynamicconfig.StringPropertyFn
 	}
 )
@@ -97,6 +98,9 @@ func (v *pinotVisibilityTripleManager) RecordWorkflowExecutionStarted(
 		func() error {
 			return v.dbVisibilityManager.RecordWorkflowExecutionStarted(ctx, request)
 		},
+		func() error {
+			return v.esVisibilityManager.RecordWorkflowExecutionStarted(ctx, request)
+		},
 		func() error {
 			return v.pinotVisibilityManager.RecordWorkflowExecutionStarted(ctx, request)
 		},
@@ -112,6 +116,9 @@ func (v *pinotVisibilityTripleManager) RecordWorkflowExecutionClosed(
 		func() error {
 			return v.dbVisibilityManager.RecordWorkflowExecutionClosed(ctx, request)
 		},
+		func() error {
+			return v.esVisibilityManager.RecordWorkflowExecutionClosed(ctx, request)
+		},
 		func() error {
 			return v.pinotVisibilityManager.RecordWorkflowExecutionClosed(ctx, request)
 		},
@@ -127,6 +134,9 @@ func (v *pinotVisibilityTripleManager) RecordWorkflowExecutionUninitialized(
 		func() error {
 			return v.dbVisibilityManager.RecordWorkflowExecutionUninitialized(ctx, request)
 		},
+		func() error {
+			return v.esVisibilityManager.RecordWorkflowExecutionUninitialized(ctx, request)
+		},
 		func() error {
 			return v.pinotVisibilityManager.RecordWorkflowExecutionUninitialized(ctx, request)
 		},
@@ -142,6 +152,9 @@ func (v *pinotVisibilityTripleManager) DeleteWorkflowExecution(
 		func() error {
 			return v.dbVisibilityManager.DeleteWorkflowExecution(ctx, request)
 		},
+		func() error {
+			return v.esVisibilityManager.DeleteWorkflowExecution(ctx, request)
+		},
 		func() error {
 			return v.pinotVisibilityManager.DeleteWorkflowExecution(ctx, request)
 		},
@@ -157,6 +170,9 @@ func (v *pinotVisibilityTripleManager) DeleteUninitializedWorkflowExecution(
 		func() error {
 			return v.dbVisibilityManager.DeleteUninitializedWorkflowExecution(ctx, request)
 		},
+		func() error {
+			return v.esVisibilityManager.DeleteUninitializedWorkflowExecution(ctx, request)
+		},
 		func() error {
 			return v.pinotVisibilityManager.DeleteUninitializedWorkflowExecution(ctx, request)
 		},
@@ -172,6 +188,9 @@ func (v *pinotVisibilityTripleManager) UpsertWorkflowExecution(
 		func() error {
 			return v.dbVisibilityManager.UpsertWorkflowExecution(ctx, request)
 		},
+		func() error {
+			return v.esVisibilityManager.UpsertWorkflowExecution(ctx, request)
+		},
 		func() error {
 			return v.pinotVisibilityManager.UpsertWorkflowExecution(ctx, request)
 		},
@@ -191,7 +210,7 @@ func (v *pinotVisibilityTripleManager) chooseVisibilityModeForAdmin() string {
 	}
 }
 
-func (v *pinotVisibilityTripleManager) chooseVisibilityManagerForWrite(ctx context.Context, dbVisFunc, pinotVisFunc func() error) error {
+func (v *pinotVisibilityTripleManager) chooseVisibilityManagerForWrite(ctx context.Context, dbVisFunc, esVisFunc, pinotVisFunc func() error) error {
 	var writeMode string
 	if v.writeMode != nil {
 		writeMode = v.writeMode()
@@ -228,6 +247,22 @@ func (v *pinotVisibilityTripleManager) chooseVisibilityManagerForWrite(ctx conte
 		}
 		v.logger.Warn("advanced visibility is not available to write")
 		return dbVisFunc()
+	case common.AdvacnedVisibilityWritingModeTriple:
+		if v.pinotVisibilityManager != nil && v.esVisibilityManager != nil {
+			if err := pinotVisFunc(); err != nil {
+				return err
+			}
+			if err := esVisFunc(); err != nil {
+				return err
+			}
+			if v.dbVisibilityManager != nil {
+				return dbVisFunc()
+			}
+			v.logger.Warn("basic visibility is not available to write")
+			return nil
+		}
+		v.logger.Warn("advanced visibility is not available to write")
+		return dbVisFunc()
 	default:
 		return &types.InternalServiceError{
 			Message: fmt.Sprintf("Unknown visibility writing mode: %s", writeMode),
@@ -333,6 +368,14 @@ func (v *pinotVisibilityTripleManager) chooseVisibilityManagerForRead(domain str
 			v.logger.Warn("domain is configured to read from advanced visibility(Pinot based) but it's not available, fall back to basic visibility",
 				tag.WorkflowDomainName(domain))
 		}
+	} else if v.readModeIsFromES(domain) {
+		if v.esVisibilityManager != nil {
+			visibilityMgr = v.esVisibilityManager
+		} else {
+			visibilityMgr = v.dbVisibilityManager
+			v.logger.Warn("domain is configured to read from advanced visibility(ElasticSearch based) but it's not available, fall back to basic visibility",
+				tag.WorkflowDomainName(domain))
+		}
 	} else {
 		if v.dbVisibilityManager != nil {
 			visibilityMgr = v.dbVisibilityManager

config/dynamicconfig/development_pinot.yaml

@@ -1,7 +1,7 @@
 frontend.enableClientVersionCheck:
   - value: true
 system.advancedVisibilityWritingMode:
-  - value: "on"
+  - value: "triple"
 system.enableReadVisibilityFromES:
   - value: true
 system.enableReadVisibilityFromPinot: