config(ticdc): add integrity configs to control row level checksum fu…

…nctionality (#8778)

close #8776

cdc/api/v2/model.go

@@ -169,6 +169,7 @@ type ReplicaConfig struct {
 	Sink       *SinkConfig                `json:"sink"`
 	Consistent *ConsistentConfig          `json:"consistent"`
 	Scheduler  *ChangefeedSchedulerConfig `json:"scheduler"`
+	Integrity  *IntegrityConfig           `json:"integrity"`
 }
 
 // ToInternalReplicaConfig coverts *v2.ReplicaConfig into *config.ReplicaConfig
@@ -294,6 +295,12 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 			WriteKeyThreshold:      c.Scheduler.WriteKeyThreshold,
 		}
 	}
+	if c.Integrity != nil {
+		res.Integrity = &config.IntegrityConfig{
+			IntegrityCheckLevel:   c.Integrity.IntegrityCheckLevel,
+			CorruptionHandleLevel: c.Integrity.CorruptionHandleLevel,
+		}
+	}
 	return res
 }
 
@@ -414,6 +421,14 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 			WriteKeyThreshold:      cloned.Scheduler.WriteKeyThreshold,
 		}
 	}
+
+	if cloned.Integrity != nil {
+		res.Integrity = &IntegrityConfig{
+			IntegrityCheckLevel:   cloned.Integrity.IntegrityCheckLevel,
+			CorruptionHandleLevel: cloned.Integrity.CorruptionHandleLevel,
+		}
+	}
+
 	return res
 }
 
@@ -575,6 +590,13 @@ type ChangefeedSchedulerConfig struct {
 	WriteKeyThreshold int `toml:"write_key_threshold" json:"write_key_threshold"`
 }
 
+// IntegrityConfig is the config for integrity check
+// This is a duplicate of config.IntegrityConfig
+type IntegrityConfig struct {
+	IntegrityCheckLevel   string `json:"integrity_check_level"`
+	CorruptionHandleLevel string `json:"corruption_handle_level"`
+}
+
 // EtcdData contains key/value pair of etcd data
 type EtcdData struct {
 	Key   string `json:"key,omitempty"`

cdc/api/v2/model_test.go

@@ -67,13 +67,18 @@ var defaultAPIConfig = &ReplicaConfig{
 		WriteKeyThreshold: config.GetDefaultReplicaConfig().
 			Scheduler.WriteKeyThreshold,
 	},
+	Integrity: &IntegrityConfig{
+		IntegrityCheckLevel:   config.GetDefaultReplicaConfig().Integrity.IntegrityCheckLevel,
+		CorruptionHandleLevel: config.GetDefaultReplicaConfig().Integrity.CorruptionHandleLevel,
+	},
 }
 
 func TestDefaultReplicaConfig(t *testing.T) {
 	t.Parallel()
 	require.Equal(t, defaultAPIConfig, GetDefaultReplicaConfig())
 	cfg := GetDefaultReplicaConfig()
 	require.NotNil(t, cfg.Scheduler)
+	require.NotNil(t, cfg.Integrity)
 	cfg2 := cfg.toInternalReplicaConfigWithOriginConfig(&config.ReplicaConfig{})
 	require.Equal(t, config.GetDefaultReplicaConfig(), cfg2)
 	cfg3 := ToAPIReplicaConfig(config.GetDefaultReplicaConfig())

cdc/model/changefeed.go

@@ -281,6 +281,10 @@ func (info *ChangeFeedInfo) VerifyAndComplete() error {
 		info.Config.Scheduler = defaultConfig.Scheduler
 	}
 
+	if info.Config.Integrity == nil {
+		info.Config.Integrity = defaultConfig.Integrity
+	}
+
 	return nil
 }
 

docs/swagger/docs.go

@@ -2008,6 +2008,17 @@ var doc = `{
                 }
             }
         },
+        "v2.IntegrityConfig": {
+            "type": "object",
+            "properties": {
+                "corruption_handle_level": {
+                    "type": "string"
+                },
+                "integrity_check_level": {
+                    "type": "string"
+                }
+            }
+        },
         "v2.LogLevelReq": {
             "type": "object",
             "properties": {
@@ -2080,6 +2091,9 @@ var doc = `{
                 "ignore_ineligible_table": {
                     "type": "boolean"
                 },
+                "integrity": {
+                    "$ref": "#/definitions/v2.IntegrityConfig"
+                },
                 "memory_quota": {
                     "type": "integer"
                 },

docs/swagger/swagger.json

@@ -1989,6 +1989,17 @@
                 }
             }
         },
+        "v2.IntegrityConfig": {
+            "type": "object",
+            "properties": {
+                "corruption_handle_level": {
+                    "type": "string"
+                },
+                "integrity_check_level": {
+                    "type": "string"
+                }
+            }
+        },
         "v2.LogLevelReq": {
             "type": "object",
             "properties": {
@@ -2061,6 +2072,9 @@
                 "ignore_ineligible_table": {
                     "type": "boolean"
                 },
+                "integrity": {
+                    "$ref": "#/definitions/v2.IntegrityConfig"
+                },
                 "memory_quota": {
                     "type": "integer"
                 },

docs/swagger/swagger.yaml

@@ -480,6 +480,13 @@ definitions:
           type: string
         type: array
     type: object
+  v2.IntegrityConfig:
+    properties:
+      corruption_handle_level:
+        type: string
+      integrity_check_level:
+        type: string
+    type: object
   v2.LogLevelReq:
     properties:
       log_level:
@@ -527,6 +534,8 @@ definitions:
         type: boolean
       ignore_ineligible_table:
         type: boolean
+      integrity:
+        $ref: '#/definitions/v2.IntegrityConfig'
       memory_quota:
         type: integer
       mounter:

pkg/config/config_test_data.go

@@ -65,7 +65,11 @@ const (
   "scheduler": {
     "enable-table-across-nodes": false,
     "region-threshold": 100000
-  }
+  },
+  "integrity": {
+    "integrity-check-level": "none",
+    "corruption-handle-level": "warn"
+ }
 }`
 
 	testCfgTestServerConfigMarshal = `{
@@ -204,7 +208,12 @@ const (
     "enable-table-across-nodes": true,
     "region-per-span": 0,
     "region-threshold": 100001,
-    "write-key-threshold": 100001
+    "write-key-threshold": 100001,
+    "region-per-span": 0
+  },
+  "integrity": {
+    "integrity-check-level": "none",
+    "corruption-handle-level": "warn"
   }
 }`
 
@@ -263,6 +272,10 @@ const (
     "enable-table-across-nodes": true,
     "region-threshold": 100001,
     "write-key-threshold": 100001
+  },
+  "integrity": {
+    "integrity-check-level": "none",
+    "corruption-handle-level": "warn"
   }
 }`
 )

pkg/config/integrity.go

@@ -0,0 +1,71 @@
+// Copyright 2021 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package config
+
+import (
+	"github.com/pingcap/log"
+	cerror "github.com/pingcap/tiflow/pkg/errors"
+	"go.uber.org/zap"
+)
+
+// IntegrityConfig represents integrity check config for a changefeed.
+type IntegrityConfig struct {
+	IntegrityCheckLevel   string `toml:"integrity-check-level" json:"integrity-check-level"`
+	CorruptionHandleLevel string `toml:"corruption-handle-level" json:"corruption-handle-level"`
+}
+
+const (
+	// IntegrityCheckLevelNone means no integrity check, the default value.
+	IntegrityCheckLevelNone string = "none"
+	// IntegrityCheckLevelCorrectness means check each row data correctness.
+	IntegrityCheckLevelCorrectness string = "correctness"
+)
+
+const (
+	// CorruptionHandleLevelWarn is the default value,
+	// log the corrupted event, and mark it as corrupted and send it to the downstream.
+	CorruptionHandleLevelWarn string = "warn"
+	// CorruptionHandleLevelError means log the corrupted event, and then stopped the changefeed.
+	CorruptionHandleLevelError string = "error"
+)
+
+// Validate the integrity config.
+func (c *IntegrityConfig) Validate() error {
+	if c.IntegrityCheckLevel != IntegrityCheckLevelNone &&
+		c.IntegrityCheckLevel != IntegrityCheckLevelCorrectness {
+		return cerror.ErrInvalidReplicaConfig.GenWithStackByArgs()
+	}
+	if c.CorruptionHandleLevel != CorruptionHandleLevelWarn &&
+		c.CorruptionHandleLevel != CorruptionHandleLevelError {
+		return cerror.ErrInvalidReplicaConfig.GenWithStackByArgs()
+	}
+
+	if c.Enabled() {
+		log.Info("integrity check is enabled",
+			zap.Any("integrityCheckLevel", c.IntegrityCheckLevel),
+			zap.Any("corruptionHandleLevel", c.CorruptionHandleLevel))
+	}
+
+	return nil
+}
+
+// Enabled returns true if the integrity check is enabled.
+func (c *IntegrityConfig) Enabled() bool {
+	return c.IntegrityCheckLevel == IntegrityCheckLevelCorrectness
+}
+
+// ErrorHandle returns true if the corruption handle level is error.
+func (c *IntegrityConfig) ErrorHandle() bool {
+	return c.CorruptionHandleLevel == CorruptionHandleLevelError
+}

pkg/config/replica_config.go

@@ -73,6 +73,10 @@ var defaultReplicaConfig = &ReplicaConfig{
 		RegionThreshold:        100_000,
 		WriteKeyThreshold:      0,
 	},
+	Integrity: &IntegrityConfig{
+		IntegrityCheckLevel:   IntegrityCheckLevelNone,
+		CorruptionHandleLevel: CorruptionHandleLevelWarn,
+	},
 }
 
 // GetDefaultReplicaConfig returns the default replica config.
@@ -114,6 +118,7 @@ type replicaConfig struct {
 	Consistent         *ConsistentConfig `toml:"consistent" json:"consistent"`
 	// Scheduler is the configuration for scheduler.
 	Scheduler *ChangefeedSchedulerConfig `toml:"scheduler" json:"scheduler"`
+	Integrity *IntegrityConfig           `toml:"integrity" json:"integrity"`
 }
 
 // Marshal returns the json marshal format of a ReplicationConfig
@@ -215,6 +220,12 @@ func (c *ReplicaConfig) ValidateAndAdjust(sinkURI *url.URL) error {
 		c.Scheduler.EnableTableAcrossNodes = false
 	}
 
+	if c.Integrity != nil {
+		if err := c.Integrity.Validate(); err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 

pkg/orchestrator/reactor_state_test.go

@@ -126,6 +126,7 @@ func TestChangefeedStateUpdate(t *testing.T) {
 						Sink:             &config.SinkConfig{Protocol: "open-protocol"},
 						Consistent:       &config.ConsistentConfig{Level: "normal", Storage: "local"},
 						Scheduler:        config.GetDefaultReplicaConfig().Scheduler,
+						Integrity:        config.GetDefaultReplicaConfig().Integrity,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713, ResolvedTs: 421980720003809281},
@@ -175,6 +176,7 @@ func TestChangefeedStateUpdate(t *testing.T) {
 						Sink:             &config.SinkConfig{Protocol: "open-protocol"},
 						Consistent:       &config.ConsistentConfig{Level: "normal", Storage: "local"},
 						Scheduler:        config.GetDefaultReplicaConfig().Scheduler,
+						Integrity:        config.GetDefaultReplicaConfig().Integrity,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713, ResolvedTs: 421980720003809281},
@@ -229,6 +231,7 @@ func TestChangefeedStateUpdate(t *testing.T) {
 						Sink:             &config.SinkConfig{Protocol: "open-protocol"},
 						Consistent:       &config.ConsistentConfig{Level: "normal", Storage: "local"},
 						Scheduler:        config.GetDefaultReplicaConfig().Scheduler,
+						Integrity:        config.GetDefaultReplicaConfig().Integrity,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713, ResolvedTs: 421980720003809281},
@@ -318,6 +321,7 @@ func TestPatchInfo(t *testing.T) {
 			Sink:       defaultConfig.Sink,
 			Consistent: defaultConfig.Consistent,
 			Scheduler:  defaultConfig.Scheduler,
+			Integrity:  defaultConfig.Integrity,
 		},
 	})
 	state.PatchInfo(func(info *model.ChangeFeedInfo) (*model.ChangeFeedInfo, bool, error) {
@@ -335,6 +339,7 @@ func TestPatchInfo(t *testing.T) {
 			Sink:       defaultConfig.Sink,
 			Consistent: defaultConfig.Consistent,
 			Scheduler:  defaultConfig.Scheduler,
+			Integrity:  defaultConfig.Integrity,
 		},
 	})
 	state.PatchInfo(func(info *model.ChangeFeedInfo) (*model.ChangeFeedInfo, bool, error) {

tests/integration_tests/api_v2/cases.go

@@ -100,6 +100,10 @@ var customReplicaConfig = &ReplicaConfig{
 		EnableTableAcrossNodes: false,
 		RegionThreshold:        13,
 	},
+	Integrity: &IntegrityConfig{
+		IntegrityCheckLevel:   "none",
+		CorruptionHandleLevel: "warn",
+	},
 }
 
 // defaultReplicaConfig check if the default values is changed
@@ -139,6 +143,10 @@ var defaultReplicaConfig = &ReplicaConfig{
 		EnableTableAcrossNodes: false,
 		RegionThreshold:        100_000,
 	},
+	Integrity: &IntegrityConfig{
+		IntegrityCheckLevel:   "none",
+		CorruptionHandleLevel: "warn",
+	},
 }
 
 func testStatus(ctx context.Context, client *CDCRESTClient) error {

tests/integration_tests/api_v2/model.go

@@ -180,6 +180,7 @@ type ReplicaConfig struct {
 	Sink       *SinkConfig                `json:"sink"`
 	Consistent *ConsistentConfig          `json:"consistent"`
 	Scheduler  *ChangefeedSchedulerConfig `json:"scheduler"`
+	Integrity  *IntegrityConfig           `json:"integrity"`
 }
 
 // FilterConfig represents filter config for a changefeed
@@ -291,6 +292,13 @@ type ChangefeedSchedulerConfig struct {
 	WriteKeyThreshold int `toml:"write_key_threshold" json:"write_key_threshold"`
 }
 
+// IntegrityConfig is the config for integrity check
+// This is a duplicate of config.IntegrityConfig
+type IntegrityConfig struct {
+	IntegrityCheckLevel   string `json:"integrity_check_level"`
+	CorruptionHandleLevel string `json:"corruption_handle_level"`
+}
+
 // ChangeFeedInfo describes the detail of a ChangeFeed
 type ChangeFeedInfo struct {
 	UpstreamID uint64    `json:"upstream_id,omitempty"`