Skip to content

log(all): add configurable TiCDC sensitive data redaction #12270

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Tema wants to merge 1 commit into
base: master
Choose a base branch
from
Open

log(all): add configurable TiCDC sensitive data redaction #12270

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion cdc/api/middleware/middleware.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import (
"github.com/pingcap/tiflow/cdc/model"
"github.com/pingcap/tiflow/pkg/config"
"github.com/pingcap/tiflow/pkg/errors"
"github.com/pingcap/tiflow/pkg/logutil"
"github.com/pingcap/tiflow/pkg/upstream"
"go.uber.org/zap"
)
Expand Down Expand Up @@ -52,7 +53,7 @@ func LogMiddleware() gin.HandlerFunc {
zap.Int("status", c.Writer.Status()),
zap.String("method", c.Request.Method),
zap.String("path", path),
zap.String("query", query),
logutil.ZapRedactString("query", query),
zap.String("ip", c.ClientIP()),
zap.String("user-agent", c.Request.UserAgent()), zap.String("client-version", version),
zap.String("username", user),
Expand Down
33 changes: 18 additions & 15 deletions cdc/entry/mounter.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ import (
cerror "github.com/pingcap/tiflow/pkg/errors"
pfilter "github.com/pingcap/tiflow/pkg/filter"
"github.com/pingcap/tiflow/pkg/integrity"
"github.com/pingcap/tiflow/pkg/logutil"
"github.com/prometheus/client_golang/prometheus"
"go.uber.org/zap"
)
Expand Down Expand Up @@ -134,7 +135,9 @@ func (m *mounter) DecodeEvent(ctx context.Context, event *model.PolymorphicEvent

func (m *mounter) unmarshalAndMountRowChanged(ctx context.Context, raw *model.RawKVEntry) (*model.RowChangedEvent, error) {
if !bytes.HasPrefix(raw.Key, tablePrefix) {
log.Error("unexpected key prefix found in row kv entry", zap.String("key", hex.EncodeToString(raw.Key)), zap.Any("eventCommitTs", raw.CRTs), zap.Any("eventStartTs", raw.StartTs))
log.Error("unexpected key prefix found in row kv entry",
logutil.ZapRedactString("key", hex.EncodeToString(raw.Key)),
zap.Any("eventCommitTs", raw.CRTs), zap.Any("eventStartTs", raw.StartTs))
return nil, nil
}
// checksumKey is only used to calculate raw checksum if necessary.
Expand All @@ -147,7 +150,7 @@ func (m *mounter) unmarshalAndMountRowChanged(ctx context.Context, raw *model.Ra
log.Warn("empty value and old value",
zap.String("namespace", m.changefeedID.Namespace),
zap.String("changefeed", m.changefeedID.ID),
zap.Any("row", raw))
logutil.ZapRedactAny("row", raw))
}
baseInfo := baseKVEntry{
StartTs: raw.StartTs,
Expand Down Expand Up @@ -178,7 +181,7 @@ func (m *mounter) unmarshalAndMountRowChanged(ctx context.Context, raw *model.Ra
}
log.Error("can not found table schema",
zap.Uint64("ts", raw.CRTs),
zap.String("key", hex.EncodeToString(raw.Key)),
logutil.ZapRedactString("key", hex.EncodeToString(raw.Key)),
zap.Int64("tableID", physicalTableID))
return nil, cerror.ErrSnapshotTableNotFound.GenWithStackByArgs(physicalTableID)
}
Expand Down Expand Up @@ -428,7 +431,7 @@ func (m *mounter) verifyColumnChecksum(
if err != nil {
log.Error("failed to calculate the checksum",
zap.Uint32("first", first), zap.Any("columnInfos", columnInfos),
zap.Any("rawColumns", rawColumns), zap.Error(err))
logutil.ZapRedactAny("rawColumns", rawColumns), zap.Error(err))
return 0, false, err
}

Expand All @@ -445,15 +448,15 @@ func (m *mounter) verifyColumnChecksum(
if !skipFail {
log.Error("cannot found the extra checksum, the first checksum mismatched",
zap.Uint32("checksum", checksum), zap.Uint32("first", first), zap.Uint32("extra", extra),
zap.Any("columnInfos", columnInfos), zap.Any("rawColumns", rawColumns), zap.Any("tz", m.tz))
zap.Any("columnInfos", columnInfos), logutil.ZapRedactAny("rawColumns", rawColumns), zap.Any("tz", m.tz))
return checksum, false, nil
}

if time.Since(m.lastSkipOldValueTime) > time.Minute {
log.Warn("checksum mismatch on the old value, "+
"this may caused by Add Column / Drop Column executed, skip verification",
zap.Uint32("checksum", checksum), zap.Uint32("first", first), zap.Uint32("extra", extra),
zap.Any("columnInfos", columnInfos), zap.Any("rawColumns", rawColumns), zap.Any("tz", m.tz))
zap.Any("columnInfos", columnInfos), logutil.ZapRedactAny("rawColumns", rawColumns), zap.Any("tz", m.tz))
m.lastSkipOldValueTime = time.Now()
}
return checksum, true, nil
Expand Down Expand Up @@ -524,7 +527,7 @@ func newDatum(value interface{}, ft types.FieldType) (types.Datum, error) {
return types.NewBytesDatum([]byte(v)), nil
}
log.Panic("unknown data type when build datum",
zap.Any("type", ft.GetType()), zap.Any("value", value), zap.Reflect("type", reflect.TypeOf(value)))
zap.Any("type", ft.GetType()), logutil.ZapRedactAny("value", value), zap.Reflect("type", reflect.TypeOf(value)))
case mysql.TypeFloat:
return types.NewFloat32Datum(value.(float32)), nil
case mysql.TypeDouble:
Expand Down Expand Up @@ -559,7 +562,7 @@ func verifyRawBytesChecksum(
datum, err := newDatum(col.Value, columnInfo.FieldType)
if err != nil {
log.Error("build datum for raw checksum calculation failed",
zap.Any("col", col), zap.Any("columnInfo", columnInfo), zap.Error(err))
logutil.ZapRedactAny("col", col), zap.Any("columnInfo", columnInfo), zap.Error(err))
return 0, false, errors.Trace(err)
}
datums = append(datums, &datum)
Expand All @@ -576,8 +579,8 @@ func verifyRawBytesChecksum(
log.Error("raw bytes checksum mismatch",
zap.Int("version", decoder.ChecksumVersion()),
zap.Uint32("expected", expected), zap.Uint32("obtained", obtained),
zap.Any("tableInfo", tableInfo), zap.Any("columns", columns),
zap.Any("handle", handle.String()), zap.Any("tz", tz))
zap.Any("tableInfo", tableInfo), logutil.ZapRedactAny("columns", columns),
logutil.ZapRedactString("handle", handle.String()), zap.Any("tz", tz))

return expected, false, nil
}
Expand Down Expand Up @@ -611,8 +614,8 @@ func (m *mounter) verifyChecksum(
expected, matched, err := verifyRawBytesChecksum(tableInfo, columns, decoder, handle, key, m.tz)
if err != nil {
log.Error("calculate raw checksum failed",
zap.Int("version", version), zap.Any("tz", m.tz), zap.Any("handle", handle.String()),
zap.Any("key", key), zap.Any("columns", columns), zap.Error(err))
zap.Int("version", version), zap.Any("tz", m.tz), logutil.ZapRedactString("handle", handle.String()),
logutil.ZapRedactStringer("key", key), logutil.ZapRedactAny("columns", columns), zap.Error(err))
return 0, false, errors.Trace(err)
}
if !matched {
Expand All @@ -621,7 +624,7 @@ func (m *mounter) verifyChecksum(
columnChecksum, err := calculateColumnChecksum(columnInfos, rawColumns, m.tz)
if err != nil {
log.Error("failed to calculate column-level checksum, after raw checksum verification passed",
zap.Any("columnsInfo", columnInfos), zap.Any("rawColumns", rawColumns),
zap.Any("columnsInfo", columnInfos), logutil.ZapRedactAny("rawColumns", rawColumns),
zap.Any("tz", m.tz), zap.Error(err))
return 0, false, errors.Trace(err)
}
Expand Down Expand Up @@ -674,7 +677,7 @@ func (m *mounter) mountRowKVEntry(
if !matched {
log.Error("previous columns checksum mismatch",
zap.Uint32("checksum", preChecksum), zap.Any("tableInfo", tableInfo),
zap.Any("preCols", preCols), zap.Any("rawCols", preRawCols))
logutil.ZapRedactAny("preCols", preCols), logutil.ZapRedactAny("rawCols", preRawCols))
if m.integrity.ErrorHandle() {
return nil, rawRow, cerror.ErrCorruptedDataMutation.
GenWithStackByArgs(m.changefeedID.Namespace, m.changefeedID.ID)
Expand All @@ -701,7 +704,7 @@ func (m *mounter) mountRowKVEntry(
if !matched {
log.Error("current columns checksum mismatch",
zap.Uint32("checksum", currentChecksum), zap.Any("tableInfo", tableInfo),
zap.Any("cols", cols), zap.Any("rawCols", rawCols))
logutil.ZapRedactAny("cols", cols), logutil.ZapRedactAny("rawCols", rawCols))
if m.integrity.ErrorHandle() {
return nil, rawRow, cerror.ErrCorruptedDataMutation.
GenWithStackByArgs(m.changefeedID.Namespace, m.changefeedID.ID)
Expand Down
3 changes: 2 additions & 1 deletion cdc/kv/matcher.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ package kv
import (
"github.com/pingcap/kvproto/pkg/cdcpb"
"github.com/pingcap/log"
"github.com/pingcap/tiflow/pkg/logutil"
"go.uber.org/zap"
)

Expand Down Expand Up @@ -106,7 +107,7 @@ func (m *matcher) matchCachedRow(initialized bool) []*cdcpb.Event_Row {
// prewrite log before initialized, a committed log with
// the same key and start-ts must have been received.
log.Info("ignore commit event without prewrite",
zap.Binary("key", cacheEntry.GetKey()),
logutil.ZapRedactAny("key", cacheEntry.GetKey()),
zap.Uint64("startTs", cacheEntry.GetStartTs()))
continue
}
Expand Down
66 changes: 66 additions & 0 deletions cdc/kv/matcher_redaction_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
// Copyright 2026 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,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package kv

import (
"bytes"
"encoding/base64"
"encoding/json"
"fmt"
"testing"

"github.com/pingcap/errors"
"github.com/pingcap/kvproto/pkg/cdcpb"
"github.com/pingcap/log"
"github.com/pingcap/tiflow/pkg/logutil/redact"
"github.com/stretchr/testify/require"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
)

func TestMatchCachedRowLogRedaction(t *testing.T) {
key := []byte("unmatched-primary-key-secret")
encodedKey := base64.StdEncoding.EncodeToString(key)
for _, mode := range []redact.RedactionMode{redact.RedactionModeOff, redact.RedactionModeOn, redact.RedactionModeMarker} {
t.Run(mode.String(), func(t *testing.T) {
originalMode := errors.RedactLogEnabled.Load()
redact.SetRedactionMode(mode)
t.Cleanup(func() { errors.RedactLogEnabled.Store(originalMode) })
var output bytes.Buffer
core := zapcore.NewCore(zapcore.NewJSONEncoder(zap.NewProductionEncoderConfig()), zapcore.AddSync(&output), zap.DebugLevel)
t.Cleanup(log.ReplaceGlobals(zap.New(core), &log.ZapProperties{Level: zap.NewAtomicLevelAt(zap.DebugLevel)}))

matcher := newMatcher()
matcher.cacheCommitRow(&cdcpb.Event_Row{StartTs: 1, CommitTs: 2, Key: key})
require.Empty(t, matcher.matchCachedRow(true))
require.Empty(t, matcher.cachedCommit)
var entry map[string]interface{}
require.NoError(t, json.Unmarshal(output.Bytes(), &entry))
require.Equal(t, "ignore commit event without prewrite", entry["msg"])
require.Equal(t, float64(1), entry["startTs"])
switch mode {
case redact.RedactionModeOff:
require.Equal(t, encodedKey, entry["key"])
case redact.RedactionModeOn:
require.Equal(t, "?", entry["key"])
require.NotContains(t, output.String(), string(key))
require.NotContains(t, output.String(), encodedKey)
require.NotContains(t, output.String(), fmt.Sprint(key))
case redact.RedactionModeMarker:
require.Equal(t, redact.String(fmt.Sprint(key)), entry["key"])
}
})
}
}
37 changes: 20 additions & 17 deletions cdc/kv/regionlock/region_range_lock.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@ import (
"github.com/google/btree"
"github.com/pingcap/log"
"github.com/pingcap/tiflow/cdc/processor/tablepb"
"github.com/pingcap/tiflow/pkg/logutil"
"github.com/pingcap/tiflow/pkg/logutil/redact"
"github.com/pingcap/tiflow/pkg/spanz"
"go.uber.org/zap"
)
Expand Down Expand Up @@ -94,8 +96,8 @@ func rangeLockEntryLess(a, b *rangeLockEntry) bool {
func (e *rangeLockEntry) String() string {
return fmt.Sprintf("region %v [%v, %v), version %v, %d waiters",
e.regionID,
hex.EncodeToString(e.startKey),
hex.EncodeToString(e.endKey),
redact.String(hex.EncodeToString(e.startKey)),
redact.String(hex.EncodeToString(e.endKey)),
e.regionVersion,
len(e.waiterSignalChs))
}
Expand Down Expand Up @@ -183,17 +185,17 @@ func (l *RangeLock) UnlockRange(
log.Panic("unlocking a not locked range",
zap.String("changefeed", l.changefeed),
zap.Uint64("regionID", regionID),
zap.String("startKey", hex.EncodeToString(startKey)),
zap.String("endKey", hex.EncodeToString(endKey)),
logutil.ZapRedactString("startKey", hex.EncodeToString(startKey)),
logutil.ZapRedactString("endKey", hex.EncodeToString(endKey)),
zap.Uint64("version", version))
}
if entry.regionID != regionID {
log.Panic("unlocked a range but regionID mismatch",
zap.String("changefeed", l.changefeed),
zap.Uint64("expectedRegionID", regionID),
zap.Uint64("foundRegionID", entry.regionID),
zap.String("startKey", hex.EncodeToString(startKey)),
zap.String("endKey", hex.EncodeToString(endKey)))
logutil.ZapRedactString("startKey", hex.EncodeToString(startKey)),
logutil.ZapRedactString("endKey", hex.EncodeToString(endKey)))
}
if entry != l.regionIDToLockedRanges[regionID] {
log.Panic("range lock and region id lock mismatch when trying to unlock",
Expand All @@ -209,8 +211,8 @@ func (l *RangeLock) UnlockRange(
log.Panic("unlocking region doesn't match the locked region",
zap.String("changefeed", l.changefeed),
zap.Uint64("regionID", regionID),
zap.String("startKey", hex.EncodeToString(startKey)),
zap.String("endKey", hex.EncodeToString(endKey)),
logutil.ZapRedactString("startKey", hex.EncodeToString(startKey)),
logutil.ZapRedactString("endKey", hex.EncodeToString(endKey)),
zap.Uint64("version", version),
zap.String("foundLockEntry", entry.String()))
}
Expand All @@ -235,8 +237,8 @@ func (l *RangeLock) UnlockRange(
zap.String("changefeed", l.changefeed),
zap.Uint64("lockID", l.id), zap.Uint64("regionID", entry.regionID),
zap.Uint64("resolvedTs", newResolvedTs),
zap.String("startKey", hex.EncodeToString(startKey)),
zap.String("endKey", hex.EncodeToString(endKey)))
logutil.ZapRedactString("startKey", hex.EncodeToString(startKey)),
logutil.ZapRedactString("endKey", hex.EncodeToString(endKey)))
return
}

Expand Down Expand Up @@ -437,8 +439,8 @@ func (l *RangeLock) tryLockRange(startKey, endKey []byte, regionID, regionVersio
zap.Uint64("regionID", regionID),
zap.Uint64("version", regionVersion),
zap.Uint64("resolvedTs", resolvedTs),
zap.String("startKey", hex.EncodeToString(startKey)),
zap.String("endKey", hex.EncodeToString(endKey)))
logutil.ZapRedactString("startKey", hex.EncodeToString(startKey)),
logutil.ZapRedactString("endKey", hex.EncodeToString(endKey)))

return LockRangeResult{
Status: LockRangeStatusSuccess,
Expand All @@ -450,7 +452,8 @@ func (l *RangeLock) tryLockRange(startKey, endKey []byte, regionID, regionVersio
var overlapStr []string
for _, r := range overlappedRangeLocks {
overlapStr = append(overlapStr, fmt.Sprintf("regionID: %v, ver: %v, start: %v, end: %v",
r.regionID, r.regionVersion, hex.EncodeToString(r.startKey), hex.EncodeToString(r.endKey))) // DEBUG
r.regionID, r.regionVersion,
redact.String(hex.EncodeToString(r.startKey)), redact.String(hex.EncodeToString(r.endKey)))) // DEBUG
}

// Check if the current acuqiring range is stale,
Expand All @@ -471,8 +474,8 @@ func (l *RangeLock) tryLockRange(startKey, endKey []byte, regionID, regionVersio
log.Info("try lock range staled",
zap.String("changefeed", l.changefeed),
zap.Uint64("lockID", l.id), zap.Uint64("regionID", regionID),
zap.String("startKey", hex.EncodeToString(startKey)),
zap.String("endKey", hex.EncodeToString(endKey)),
logutil.ZapRedactString("startKey", hex.EncodeToString(startKey)),
logutil.ZapRedactString("endKey", hex.EncodeToString(endKey)),
zap.Strings("allOverlapping", overlapStr)) // DEBUG

for _, r := range overlappedRangeLocks {
Expand Down Expand Up @@ -512,8 +515,8 @@ func (l *RangeLock) tryLockRange(startKey, endKey []byte, regionID, regionVersio
log.Info("lock range blocked",
zap.String("changefeed", l.changefeed),
zap.Uint64("lockID", l.id), zap.Uint64("regionID", regionID),
zap.String("startKey", hex.EncodeToString(startKey)),
zap.String("endKey", hex.EncodeToString(endKey)),
logutil.ZapRedactString("startKey", hex.EncodeToString(startKey)),
logutil.ZapRedactString("endKey", hex.EncodeToString(endKey)),
zap.Strings("blockedBy", overlapStr)) // DEBUG

return LockRangeResult{
Expand Down
Loading