mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-26 23:10:44 +01:00
filer,pb,mount: keep per-chunk cipher keys out of logs (#12085)
* filer: format ChunkView without its cipher key readChunkSliceAt logs a failed chunk fetch with %+v, which printed the per-chunk AES-256-GCM key into error-level logs on every failed read. Give ChunkView a String() that omits CipherKey so %v and %+v stay safe. * pb: scrub cipher keys when an entry must be logged whole ScrubCipherKeys clones an Entry with chunk CipherKey fields cleared, so a warning that needs the full entry still cannot leak the keys. * log entry identifiers instead of whole metadata messages Several log lines formatted whole protobuf messages — SubscribeMetadataResponse, EventNotification, Entry-carrying requests — whose chunks include cipher_key under -encryptVolumeData, so the keys escaped into logs (some at default verbosity). Log the directory, entry name and event timestamp instead; where a chunk list was printed, print file ids via FileIdsString. * log queue: unwrap SubscribeMetadataResponse and never print raw protos The fallback printed any message with %+v, so a SubscribeMetadataResponse carrying entries would still put cipher keys in the log. Unwrap its event notifications (batched events included) and reduce unknown messages to their type name. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * remote sync logs: name the event's real path Rename targets logged under resp.Directory while the entry lives under NewParentPath; route them through MetadataEventTargetDirectory. Sync failure logs printed only the directory, so same-dir failures shared an identifier — log the source and destination paths instead. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * log queue: log the entry's destination path on moves A cross-directory rename printed "file" -> "file" under the source path, hiding where the entry landed. Use NewParentPath so the log shows the destination directory next to the new name. * admin: install the go.mod-pinned templ, not @latest install-templ fetched templ@latest, which now emits templruntime.NewAttributeExpression - a symbol absent from the runtime pinned in go.mod - so every fresh 'make' regenerated uncompilable _templ.go files. Install the go.mod version and reinstall when the binary on PATH reports a different one. * admin: drop stray Makefile change belonging to the templ pin PR Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * notification: join the move destination path in log queue output A NewParentPath of "/" printed //name for moves into the root; path.Join keeps a single leading slash. * log queue: skip a nil event notification instead of logging an empty move Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * pb: tolerate nil chunk entries while scrubbing cipher keys for logs Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * subscription and backup failure logs name the event's file The process/marshal failure logs printed only the event directory and timestamp, so a failed sync left operators guessing which file failed. Use MetadataEventSourceFullPath/MetadataEventTargetFullPath so each failure names the source and target paths without printing the entries. * replication: identify entry-less events without printing the proto Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * s3: scrub cipher keys from the bucket-metadata debug dump buildBucketMetadata marshals the whole filer entry to JSON for a V(3) log line; a chunked entry would emit each chunk's cipher_key base64. Marshal the scrubbed copy instead. --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
1 parent
c27d7fe5a4
commit
3abf8b2572
26 files changed
+145
-44
No files matched your search
@@ -235,7 +235,7 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
|
||||
inner := processEventFn
|
||||
processEventFn = func(resp *filer_pb.SubscribeMetadataResponse) error {
|
||||
if err := inner(resp); err != nil {
|
||||
glog.Errorf("process %v: %v", resp, err)
|
||||
glog.Errorf("process %s -> %s ts_ns=%d: %v", filer_pb.MetadataEventSourceFullPath(resp), filer_pb.MetadataEventTargetFullPath(resp), resp.TsNs, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -254,9 +254,9 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
glog.V(2).Infof("create: %+v", resp)
|
||||
glog.V(2).Infof("create: %s/%s ts_ns=%d", filer_pb.MetadataEventTargetDirectory(resp), message.GetNewEntry().GetName(), resp.TsNs)
|
||||
if !shouldSendToRemote(message.NewEntry) {
|
||||
glog.V(2).Infof("skipping creating: %+v", resp)
|
||||
glog.V(2).Infof("skipping creating: %s/%s ts_ns=%d", filer_pb.MetadataEventTargetDirectory(resp), message.GetNewEntry().GetName(), resp.TsNs)
|
||||
return nil
|
||||
}
|
||||
// Rewrite internal versioning paths to the original S3 key
|
||||
@@ -320,7 +320,7 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
glog.V(2).Infof("delete: %+v", resp)
|
||||
glog.V(2).Infof("delete: %s/%s ts_ns=%d", resp.Directory, message.GetOldEntry().GetName(), resp.TsNs)
|
||||
dest := toRemoteStorageLocation(bucket, util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation)
|
||||
if message.OldEntry.IsDirectory {
|
||||
glog.V(0).Infof("rmdir %s", remote_storage.FormatLocation(dest))
|
||||
@@ -361,7 +361,7 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour
|
||||
newBucket, newRemoteStorageMountLocation, newRemoteStorage, newOk := option.detectBucketInfo(message.NewParentPath)
|
||||
if oldOk && newOk {
|
||||
if !shouldSendToRemote(message.NewEntry) {
|
||||
glog.V(2).Infof("skipping updating: %+v", resp)
|
||||
glog.V(2).Infof("skipping updating: %s/%s ts_ns=%d", filer_pb.MetadataEventTargetDirectory(resp), message.GetNewEntry().GetName(), resp.TsNs)
|
||||
return nil
|
||||
}
|
||||
client, err := remote_storage.GetRemoteStorage(oldRemoteStorage)
|
||||
@@ -375,7 +375,7 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour
|
||||
return nil
|
||||
}
|
||||
if message.OldEntry.RemoteEntry != nil && filer.IsSameData(message.OldEntry, message.NewEntry) {
|
||||
glog.V(2).Infof("update meta: %+v", resp)
|
||||
glog.V(2).Infof("update meta: %s/%s ts_ns=%d", filer_pb.MetadataEventTargetDirectory(resp), message.GetNewEntry().GetName(), resp.TsNs)
|
||||
oldDest := toRemoteStorageLocation(oldBucket, util.NewFullPath(resp.Directory, message.OldEntry.Name), oldRemoteStorageMountLocation)
|
||||
updated, err := client.UpdateFileMetadata(oldDest, message.OldEntry, message.NewEntry)
|
||||
if err != nil {
|
||||
@@ -425,7 +425,7 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour
|
||||
}
|
||||
if newOk {
|
||||
if !shouldSendToRemote(message.NewEntry) {
|
||||
glog.V(2).Infof("skipping updating: %+v", resp)
|
||||
glog.V(2).Infof("skipping updating: %s/%s ts_ns=%d", filer_pb.MetadataEventTargetDirectory(resp), message.GetNewEntry().GetName(), resp.TsNs)
|
||||
return nil
|
||||
}
|
||||
client, err := remote_storage.GetRemoteStorage(newRemoteStorage)
|
||||
|
||||
@@ -260,9 +260,9 @@ func processCreateEvent(
|
||||
if !filer.HasData(message.NewEntry) && !(mustSend && !isDroppedLocalCopy(message.NewEntry)) {
|
||||
return nil
|
||||
}
|
||||
glog.V(2).Infof("create: %+v", resp)
|
||||
glog.V(2).Infof("create: %s/%s ts_ns=%d", filer_pb.MetadataEventTargetDirectory(resp), message.GetNewEntry().GetName(), resp.TsNs)
|
||||
if !mustSend && !shouldSendToRemote(message.NewEntry) {
|
||||
glog.V(2).Infof("skipping creating: %+v", resp)
|
||||
glog.V(2).Infof("skipping creating: %s/%s ts_ns=%d", filer_pb.MetadataEventTargetDirectory(resp), message.GetNewEntry().GetName(), resp.TsNs)
|
||||
return nil
|
||||
}
|
||||
// Rewrite internal versioning paths to the original S3 key
|
||||
@@ -332,7 +332,7 @@ func processDeleteEvent(
|
||||
glog.V(2).Infof("skipping remote delete of metadata-only delete: %s/%s", resp.Directory, message.OldEntry.Name)
|
||||
return nil
|
||||
}
|
||||
glog.V(2).Infof("delete: %+v", resp)
|
||||
glog.V(2).Infof("delete: %s/%s ts_ns=%d", resp.Directory, message.GetOldEntry().GetName(), resp.TsNs)
|
||||
dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation)
|
||||
if !message.OldEntry.IsDirectory {
|
||||
// a deleted entry whose rename copy was still pending read its
|
||||
@@ -456,7 +456,7 @@ func processUpdateEvent(
|
||||
return fallback
|
||||
}
|
||||
if proto.Equal(oldDest, dest) && !shouldSendToRemote(message.NewEntry) {
|
||||
glog.V(2).Infof("skipping updating: %+v", resp)
|
||||
glog.V(2).Infof("skipping updating: %s/%s ts_ns=%d", filer_pb.MetadataEventTargetDirectory(resp), message.GetNewEntry().GetName(), resp.TsNs)
|
||||
return nil
|
||||
}
|
||||
if message.NewEntry.IsDirectory {
|
||||
@@ -465,14 +465,14 @@ func processUpdateEvent(
|
||||
if isMetadataOnlyUpdate(resp.Directory, message) {
|
||||
remoteEntry, err := liveRemoteEntry(filerClient, message.NewParentPath, message.NewEntry)
|
||||
if errors.Is(err, filer_pb.ErrNotFound) {
|
||||
glog.V(2).Infof("skipping updating deleted entry: %+v", resp)
|
||||
glog.V(2).Infof("skipping updating deleted entry: %s/%s ts_ns=%d", filer_pb.MetadataEventTargetDirectory(resp), message.GetNewEntry().GetName(), resp.TsNs)
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if remoteEntry != nil {
|
||||
glog.V(2).Infof("update meta: %+v", resp)
|
||||
glog.V(2).Infof("update meta: %s/%s ts_ns=%d", filer_pb.MetadataEventTargetDirectory(resp), message.GetNewEntry().GetName(), resp.TsNs)
|
||||
updated, err := client.UpdateFileMetadata(dest, message.OldEntry, remoteWriteEntry(message.NewEntry, storageClass))
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -494,10 +494,10 @@ func processUpdateEvent(
|
||||
// the entry as dirty when the remote mtime shares its second. The
|
||||
// content is the object at this very key, and uploading the chunkless
|
||||
// entry would replace it with zero bytes.
|
||||
glog.V(2).Infof("skipping update of remote-only entry: %+v", resp)
|
||||
glog.V(2).Infof("skipping update of remote-only entry: %s/%s ts_ns=%d", filer_pb.MetadataEventTargetDirectory(resp), message.GetNewEntry().GetName(), resp.TsNs)
|
||||
return nil
|
||||
}
|
||||
glog.V(2).Infof("update: %+v", resp)
|
||||
glog.V(2).Infof("update: %s/%s ts_ns=%d", filer_pb.MetadataEventTargetDirectory(resp), message.GetNewEntry().GetName(), resp.TsNs)
|
||||
if !proto.Equal(oldDest, dest) {
|
||||
// A renamed entry that holds no local data now (the snapshot was
|
||||
// remote-only, or remote.uncache ran since) has its content only as a
|
||||
|
||||
@@ -683,7 +683,7 @@ func genProcessFunction(sourcePath string, targetPath string, excludePaths []str
|
||||
}
|
||||
|
||||
if debug {
|
||||
glog.V(0).Infof("received %v", resp)
|
||||
glog.V(0).Infof("received %s ts_ns=%d", resp.Directory, resp.TsNs)
|
||||
}
|
||||
|
||||
if isMultipartUploadDir(resp.Directory + "/") {
|
||||
|
||||
@@ -479,7 +479,7 @@ func (t *MetadataProcessor) AddSyncJob(resp *filer_pb.SubscribeMetadataResponse)
|
||||
if resp.TsNs < t.oldestFailedTsNs {
|
||||
t.oldestFailedTsNs = resp.TsNs
|
||||
}
|
||||
glog.Errorf("process %v: %v", resp, jobErr)
|
||||
glog.Errorf("process %s -> %s ts_ns=%d: %v", jobPaths.path, jobPaths.newPath, resp.TsNs, jobErr)
|
||||
} else if _, recorded := t.failedTs[failedKey]; !recorded {
|
||||
if len(t.failedTs) >= maxFailedSyncEvents {
|
||||
t.failedSticky = true
|
||||
@@ -487,14 +487,14 @@ func (t *MetadataProcessor) AddSyncJob(resp *filer_pb.SubscribeMetadataResponse)
|
||||
if resp.TsNs < t.oldestFailedTsNs {
|
||||
t.oldestFailedTsNs = resp.TsNs
|
||||
}
|
||||
glog.Warningf("process %v: %v; over %d unresolved failures, pinning sync offset at %v until restart", resp, jobErr, maxFailedSyncEvents, time.Unix(0, t.oldestFailedTsNs))
|
||||
glog.Warningf("process %s -> %s ts_ns=%d: %v; over %d unresolved failures, pinning sync offset at %v until restart", jobPaths.path, jobPaths.newPath, resp.TsNs, jobErr, maxFailedSyncEvents, time.Unix(0, t.oldestFailedTsNs))
|
||||
} else {
|
||||
t.failedTs[failedKey] = struct{}{}
|
||||
if t.oldestFailedTsNs == 0 || resp.TsNs < t.oldestFailedTsNs {
|
||||
t.oldestFailedTsNs = resp.TsNs
|
||||
glog.Errorf("process %v: %v; holding sync offset at %v so this event is replayed on restart", resp, jobErr, time.Unix(0, resp.TsNs))
|
||||
glog.Errorf("process %s -> %s ts_ns=%d: %v; holding sync offset at %v so this event is replayed on restart", jobPaths.path, jobPaths.newPath, resp.TsNs, jobErr, time.Unix(0, resp.TsNs))
|
||||
} else {
|
||||
glog.Errorf("process %v: %v", resp, jobErr)
|
||||
glog.Errorf("process %s -> %s ts_ns=%d: %v", jobPaths.path, jobPaths.newPath, resp.TsNs, jobErr)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -169,6 +169,12 @@ type ChunkView struct {
|
||||
ModifiedTsNs int64
|
||||
}
|
||||
|
||||
// String omits CipherKey so %-format logging can never leak a chunk's key.
|
||||
func (cv *ChunkView) String() string {
|
||||
return fmt.Sprintf("{FileId:%s OffsetInChunk:%d ViewSize:%d ViewOffset:%d ChunkSize:%d IsGzipped:%v ModifiedTsNs:%d}",
|
||||
cv.FileId, cv.OffsetInChunk, cv.ViewSize, cv.ViewOffset, cv.ChunkSize, cv.IsGzipped, cv.ModifiedTsNs)
|
||||
}
|
||||
|
||||
func (cv *ChunkView) SetStartStop(start, stop int64) {
|
||||
cv.OffsetInChunk += start - cv.ViewOffset
|
||||
cv.ViewOffset = start
|
||||
|
||||
@@ -14,6 +14,16 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
)
|
||||
|
||||
func TestChunkViewStringOmitsCipherKey(t *testing.T) {
|
||||
key := []byte("0123456789abcdef0123456789abcdef")
|
||||
cv := &ChunkView{FileId: "1,aaa", CipherKey: key}
|
||||
for _, format := range []string{"%v", "%+v", "%s"} {
|
||||
got := fmt.Sprintf(format, cv)
|
||||
assert.NotContains(t, got, string(key), "format %s", format)
|
||||
assert.Contains(t, got, cv.FileId, "format %s", format)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCompactFileChunks(t *testing.T) {
|
||||
chunks := []*filer_pb.FileChunk{
|
||||
{Offset: 10, Size: 100, FileId: "abc", ModifiedTsNs: 50},
|
||||
|
||||
@@ -675,7 +675,7 @@ func (f *Filer) deleteChunksIfNotNew(ctx context.Context, oldEntry, newEntry *En
|
||||
|
||||
toDelete, err := MinusChunks(ctx, f.MasterClient.GetLookupFileIdFunction(), oldChunks, newChunks, f.MasterClient)
|
||||
if err != nil {
|
||||
glog.ErrorfCtx(ctx, "Failed to resolve old entry chunks when delete old entry chunks. new: %s, old: %s", newChunks, oldChunks)
|
||||
glog.ErrorfCtx(ctx, "Failed to resolve old entry chunks when delete old entry chunks. new: %v, old: %v", filer_pb.FileIdsString(newChunks), filer_pb.FileIdsString(oldChunks))
|
||||
return
|
||||
}
|
||||
f.DeleteChunksNotRecursive(toDelete)
|
||||
|
||||
@@ -289,7 +289,7 @@ func (f *Filer) newMetadataEvent(oldEntry, newEntry *Entry, deleteChunks, isFrom
|
||||
func (f *Filer) logMetaEvent(ctx context.Context, event *filer_pb.SubscribeMetadataResponse) {
|
||||
data, err := proto.Marshal(event)
|
||||
if err != nil {
|
||||
glog.Errorf("failed to marshal filer_pb.SubscribeMetadataResponse %+v: %v", event, err)
|
||||
glog.Errorf("failed to marshal filer_pb.SubscribeMetadataResponse %s -> %s ts_ns=%d: %v", filer_pb.MetadataEventSourceFullPath(event), filer_pb.MetadataEventTargetFullPath(event), event.TsNs, err)
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@@ -99,7 +99,7 @@ func (f *Filer) maybeReloadFilerConfiguration(event *filer_pb.SubscribeMetadataR
|
||||
return
|
||||
}
|
||||
|
||||
glog.V(0).Infof("processing %v", event)
|
||||
glog.V(0).Infof("processing %s ts_ns=%d", event.Directory, event.TsNs)
|
||||
if entry.Name == FilerConfName {
|
||||
f.reloadFilerConfiguration(entry)
|
||||
}
|
||||
|
||||
@@ -549,7 +549,7 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
|
||||
processEventFn := func(event *filer_pb.SubscribeMetadataResponse) error {
|
||||
data, err := proto.Marshal(event)
|
||||
if err != nil {
|
||||
glog.Errorf("failed to marshal subscribed filer_pb.SubscribeMetadataResponse %+v: %v", event, err)
|
||||
glog.Errorf("failed to marshal subscribed filer_pb.SubscribeMetadataResponse %s -> %s ts_ns=%d: %v", filer_pb.MetadataEventSourceFullPath(event), filer_pb.MetadataEventTargetFullPath(event), event.TsNs, err)
|
||||
return err
|
||||
}
|
||||
dir := event.Directory
|
||||
@@ -607,8 +607,8 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
|
||||
default:
|
||||
}
|
||||
if err := processEventFn(event); err != nil {
|
||||
glog.V(0).Infof("SubscribeLocalMetadata process %v: %v", event, err)
|
||||
return fmt.Errorf("process %v: %w", event, err)
|
||||
glog.V(0).Infof("SubscribeLocalMetadata process %s -> %s ts_ns=%d: %v", filer_pb.MetadataEventSourceFullPath(event), filer_pb.MetadataEventTargetFullPath(event), event.TsNs, err)
|
||||
return fmt.Errorf("process %s -> %s ts_ns=%d: %w", filer_pb.MetadataEventSourceFullPath(event), filer_pb.MetadataEventTargetFullPath(event), event.TsNs, err)
|
||||
}
|
||||
f.onMetadataChangeEvent(event)
|
||||
lastTsNs = event.TsNs
|
||||
|
||||
@@ -134,7 +134,7 @@ func (fh *FileHandle) SetEntry(entry *filer_pb.Entry) {
|
||||
var resolveManifestErr error
|
||||
fh.entryChunkGroup, resolveManifestErr = filer.NewChunkGroup(fh.wfs.LookupFn(), fh.wfs.chunkCache, entry.Chunks, fh.wfs.option.ConcurrentReaders, fh.wfs.CacheInvalidator(), fh.wfs.manifestCache, fh.wfs.readerCacheBudget)
|
||||
if resolveManifestErr != nil {
|
||||
glog.Warningf("failed to resolve manifest chunks in %+v", entry)
|
||||
glog.Warningf("failed to resolve manifest chunks in %+v", filer_pb.ScrubCipherKeys(entry))
|
||||
}
|
||||
} else {
|
||||
glog.Fatalf("setting file handle entry to nil")
|
||||
|
||||
@@ -437,7 +437,7 @@ func (wfs *WFS) createRegularFile(ctx context.Context, dirFullPath util.FullPath
|
||||
SkipCheckParentDirectory: true,
|
||||
}
|
||||
|
||||
glog.V(1).Infof("createFile: %v", request)
|
||||
glog.V(1).Infof("createFile: %s", entryFullPath)
|
||||
resp, err := wfs.streamCreateEntry(context.Background(), request)
|
||||
if err != nil {
|
||||
glog.V(0).Infof("createFile %s: %v", entryFullPath, err)
|
||||
|
||||
@@ -481,7 +481,7 @@ func (wfs *WFS) Rename(cancel <-chan struct{}, in *fuse.RenameIn, oldName string
|
||||
func (wfs *WFS) handleRenameResponse(ctx context.Context, resp *filer_pb.StreamRenameEntryResponse, renameNewPath util.FullPath, renameNewPathLock **cluster.LiveLock) error {
|
||||
// comes from filer StreamRenameEntry, can only be create or delete entry
|
||||
|
||||
glog.V(4).Infof("dir Rename %+v", resp.EventNotification)
|
||||
glog.V(4).Infof("dir Rename %s/%s -> %s/%s", resp.Directory, resp.EventNotification.GetOldEntry().GetName(), resp.EventNotification.GetNewParentPath(), resp.EventNotification.GetNewEntry().GetName())
|
||||
|
||||
if resp.EventNotification.NewEntry != nil {
|
||||
if err := wfs.applyLocalMetadataEvent(ctx, metadataEventFromRenameResponse(resp)); err != nil {
|
||||
|
||||
@@ -27,7 +27,7 @@ func (wfs *WFS) saveEntry(path util.FullPath, entry *filer_pb.Entry) (code fuse.
|
||||
Signatures: []int32{wfs.signature},
|
||||
}
|
||||
|
||||
glog.V(1).Infof("save entry: %v", request)
|
||||
glog.V(1).Infof("save entry %s", path)
|
||||
|
||||
var resp *filer_pb.UpdateEntryResponse
|
||||
err := retryMetadataFlushIf(context.Background(), func() error {
|
||||
|
||||
@@ -1,8 +1,11 @@
|
||||
package kafka
|
||||
|
||||
import (
|
||||
"path"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/notification"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
"google.golang.org/protobuf/proto"
|
||||
)
|
||||
@@ -24,6 +27,24 @@ func (k *LogQueue) Initialize(configuration util.Configuration, prefix string) (
|
||||
|
||||
func (k *LogQueue) SendMessage(key string, message proto.Message) (err error) {
|
||||
|
||||
glog.V(0).Infof("%v: %+v", key, message)
|
||||
// entries carry per-chunk cipher keys; log names only
|
||||
if resp, ok := message.(*filer_pb.SubscribeMetadataResponse); ok {
|
||||
if notification := resp.GetEventNotification(); notification != nil {
|
||||
_ = k.SendMessage(key, notification)
|
||||
}
|
||||
for _, event := range resp.GetEvents() {
|
||||
_ = k.SendMessage(key, event)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if event, ok := message.(*filer_pb.EventNotification); ok {
|
||||
name := event.GetNewEntry().GetName()
|
||||
if parent := event.GetNewParentPath(); parent != "" {
|
||||
name = path.Join(parent, name)
|
||||
}
|
||||
glog.V(0).Infof("%v: %q -> %q", key, event.GetOldEntry().GetName(), name)
|
||||
return nil
|
||||
}
|
||||
glog.V(0).Infof("%v: %T", key, message)
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
package filer_pb
|
||||
|
||||
import "google.golang.org/protobuf/proto"
|
||||
|
||||
// ScrubCipherKeys returns a copy of entry with every chunk's CipherKey
|
||||
// cleared, so logging it cannot leak per-chunk encryption keys.
|
||||
func ScrubCipherKeys(entry *Entry) *Entry {
|
||||
if entry == nil {
|
||||
return nil
|
||||
}
|
||||
clone := proto.Clone(entry).(*Entry)
|
||||
for _, chunk := range clone.Chunks {
|
||||
if chunk != nil {
|
||||
chunk.CipherKey = nil
|
||||
}
|
||||
}
|
||||
return clone
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
package filer_pb
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestScrubCipherKeys(t *testing.T) {
|
||||
key := []byte("0123456789abcdef0123456789abcdef")
|
||||
entry := &Entry{
|
||||
Name: "f",
|
||||
Chunks: []*FileChunk{
|
||||
{FileId: "1,aaa", CipherKey: key},
|
||||
{FileId: "2,bbb"},
|
||||
},
|
||||
}
|
||||
|
||||
scrubbed := ScrubCipherKeys(entry)
|
||||
|
||||
for i, chunk := range scrubbed.Chunks {
|
||||
if len(chunk.CipherKey) != 0 {
|
||||
t.Fatalf("chunk %d still carries a cipher key", i)
|
||||
}
|
||||
}
|
||||
if len(entry.Chunks[0].CipherKey) != len(key) {
|
||||
t.Fatal("scrubbing mutated the original entry")
|
||||
}
|
||||
if got := fmt.Sprintf("%v", scrubbed); got == fmt.Sprintf("%v", entry) {
|
||||
t.Fatal("scrubbed entry prints identically to the original")
|
||||
}
|
||||
if ScrubCipherKeys(nil) != nil {
|
||||
t.Fatal("nil entry should scrub to nil")
|
||||
}
|
||||
}
|
||||
@@ -44,7 +44,7 @@ func GetEntry(ctx context.Context, filerClient FilerClient, fullFilePath util.Fu
|
||||
// glog.V(3).Infof("read %s request: %v", fullFilePath, request)
|
||||
resp, err := LookupEntry(ctx, client, request)
|
||||
if err != nil {
|
||||
glog.V(3).InfofCtx(ctx, "read %s %v: %v", fullFilePath, resp, err)
|
||||
glog.V(3).InfofCtx(ctx, "read %s: %v", fullFilePath, err)
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -337,7 +337,7 @@ func MkFile(ctx context.Context, filerClient FilerClient, parentDirectoryPath st
|
||||
|
||||
glog.V(1).InfofCtx(ctx, "create file: %s/%s", parentDirectoryPath, fileName)
|
||||
if err := CreateEntry(ctx, client, request); err != nil {
|
||||
glog.V(0).InfofCtx(ctx, "create file %v:%v", request, err)
|
||||
glog.V(0).InfofCtx(ctx, "create file %s/%s: %v", parentDirectoryPath, fileName, err)
|
||||
return fmt.Errorf("create file %s/%s: %w", parentDirectoryPath, fileName, err)
|
||||
}
|
||||
|
||||
|
||||
@@ -117,6 +117,16 @@ func (c *FileChunk) GetFileIdString() string {
|
||||
return ""
|
||||
}
|
||||
|
||||
// FileIdsString lists chunks by file id only, for logging that must not print
|
||||
// per-chunk fields like CipherKey.
|
||||
func FileIdsString(chunks []*FileChunk) []string {
|
||||
fileIds := make([]string, 0, len(chunks))
|
||||
for _, chunk := range chunks {
|
||||
fileIds = append(fileIds, chunk.GetFileIdString())
|
||||
}
|
||||
return fileIds
|
||||
}
|
||||
|
||||
func BeforeEntrySerialization(chunks []*FileChunk) {
|
||||
|
||||
for _, chunk := range chunks {
|
||||
|
||||
@@ -120,18 +120,20 @@ func makeSubscribeMetadataFunc(option *MetadataFollowOption, processEventFn Proc
|
||||
}
|
||||
|
||||
handleErr := func(resp *filer_pb.SubscribeMetadataResponse, err error) {
|
||||
sourcePath := filer_pb.MetadataEventSourceFullPath(resp)
|
||||
targetPath := filer_pb.MetadataEventTargetFullPath(resp)
|
||||
switch option.EventErrorType {
|
||||
case TrivialOnError:
|
||||
glog.Errorf("process %v: %v", resp, err)
|
||||
glog.Errorf("process %s -> %s ts_ns=%d: %v", sourcePath, targetPath, resp.TsNs, err)
|
||||
case FatalOnError:
|
||||
glog.Fatalf("process %v: %v", resp, err)
|
||||
glog.Fatalf("process %s -> %s ts_ns=%d: %v", sourcePath, targetPath, resp.TsNs, err)
|
||||
case RetryForeverOnError:
|
||||
waitTime := time.Second
|
||||
for ctx.Err() == nil {
|
||||
if err := processEventFn(resp); err == nil {
|
||||
break
|
||||
} else {
|
||||
glog.Errorf("process %v: %v", resp, err)
|
||||
glog.Errorf("process %s -> %s ts_ns=%d: %v", sourcePath, targetPath, resp.TsNs, err)
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
@@ -144,7 +146,7 @@ func makeSubscribeMetadataFunc(option *MetadataFollowOption, processEventFn Proc
|
||||
case DontLogError:
|
||||
// pass
|
||||
default:
|
||||
glog.Errorf("process %v: %v", resp, err)
|
||||
glog.Errorf("process %s -> %s ts_ns=%d: %v", sourcePath, targetPath, resp.TsNs, err)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -108,7 +108,7 @@ func (r *Replicator) Replicate(ctx context.Context, key string, message *filer_p
|
||||
return r.sink.CreateEntry(oldSinkKey, newEntry, message.Signatures)
|
||||
}
|
||||
if oldEntry == nil && newEntry == nil {
|
||||
glog.V(0).Infof("weird message %+v", message)
|
||||
glog.V(0).Infof("weird message %v: no old or new entry", key)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -321,7 +321,7 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [
|
||||
Signatures: signatures,
|
||||
}
|
||||
|
||||
glog.V(3).Infof("create: %v", request)
|
||||
glog.V(3).Infof("create: %s", key)
|
||||
if err := filer_pb.CreateEntry(context.Background(), client, request); err != nil {
|
||||
glog.V(0).Infof("create entry %s: %v", key, err)
|
||||
return fmt.Errorf("create entry %s: %w", key, err)
|
||||
@@ -360,7 +360,7 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
|
||||
return false, fmt.Errorf("lookup %s: %w", key, err)
|
||||
}
|
||||
|
||||
glog.V(4).Infof("oldEntry %+v, newEntry %+v, existingEntry: %+v", oldEntry, newEntry, existingEntry)
|
||||
glog.V(4).Infof("entry %s: old %q new %q existing %q", key, oldEntry.GetName(), newEntry.GetName(), existingEntry.GetName())
|
||||
|
||||
// the supersession checks below must look up where the incoming entry lives
|
||||
// now, which for a rename is newParentPath/newEntry.Name, not the old key.
|
||||
|
||||
@@ -94,7 +94,7 @@ func bucketOwnerAccountId(accountManager AccountManager, entry *filer_pb.Entry)
|
||||
}
|
||||
|
||||
func buildBucketMetadata(accountManager AccountManager, entry *filer_pb.Entry) *BucketMetaData {
|
||||
entryJson, _ := json.Marshal(entry)
|
||||
entryJson, _ := json.Marshal(filer_pb.ScrubCipherKeys(entry))
|
||||
glog.V(3).Infof("build bucket metadata,entry=%s", entryJson)
|
||||
bucketMetadata := &BucketMetaData{
|
||||
Name: entry.Name,
|
||||
|
||||
@@ -821,7 +821,7 @@ func (fs *FilerServer) applyRecomputeLatest(ctx context.Context, m *filer_pb.Obj
|
||||
|
||||
func (fs *FilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) {
|
||||
|
||||
glog.V(4).InfofCtx(ctx, "UpdateEntry %v", req)
|
||||
glog.V(4).InfofCtx(ctx, "UpdateEntry %s/%s", req.GetDirectory(), req.GetEntry().GetName())
|
||||
if len(req.Entry.HardLinkId) > 0 {
|
||||
glog.V(4).InfofCtx(ctx, "UpdateEntry %s/%s with HardLinkId %x counter=%d", req.Directory, req.Entry.Name, req.Entry.HardLinkId, req.Entry.HardLinkCounter)
|
||||
}
|
||||
@@ -990,7 +990,7 @@ func (fs *FilerServer) cleanupChunks(ctx context.Context, fullpath string, exist
|
||||
|
||||
func (fs *FilerServer) AppendToEntry(ctx context.Context, req *filer_pb.AppendToEntryRequest) (*filer_pb.AppendToEntryResponse, error) {
|
||||
|
||||
glog.V(4).InfofCtx(ctx, "AppendToEntry %v", req)
|
||||
glog.V(4).InfofCtx(ctx, "AppendToEntry %s/%s", req.GetDirectory(), req.GetEntryName())
|
||||
fullpath := util.NewFullPath(req.Directory, req.EntryName)
|
||||
|
||||
lockClient := cluster.NewLockClient(fs.grpcDialOption, fs.option.Host)
|
||||
|
||||
@@ -467,7 +467,7 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee
|
||||
if existingTarget != nil {
|
||||
toDelete, err := filer.MinusChunks(ctx, fs.filer.MasterClient.GetLookupFileIdFunction(), existingTarget.GetChunks(), newEntry.GetChunks(), fs.filer.MasterClient)
|
||||
if err != nil {
|
||||
glog.ErrorfCtx(ctx, "Failed to resolve overwrite target chunks during rename. new: %v, old: %v", newEntry.GetChunks(), existingTarget.GetChunks())
|
||||
glog.ErrorfCtx(ctx, "Failed to resolve overwrite target chunks during rename. new: %v, old: %v", filer_pb.FileIdsString(newEntry.GetChunks()), filer_pb.FileIdsString(existingTarget.GetChunks()))
|
||||
} else if len(toDelete) > 0 {
|
||||
// Defer chunk deletion until after CommitTransaction so that a
|
||||
// failure in any subsequent step (child moves, oldPath delete,
|
||||
|
||||
Reference in new issue
Block a user