-
Notifications
You must be signed in to change notification settings - Fork 1.7k
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.
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
Re-open the log file by filesystem notify #43359
Changes from all commits
bdfa269
6a5c99f
29c994c
f34ae03
926f68a
f14d2b6
81f0296
49ab434
686a8dd
550d75a
991c70a
4c762fb
d1ed93c
babf127
580c801
a303a8b
10042fa
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,164 @@ | ||
/* | ||
* Teleport | ||
* Copyright (C) 2024 Gravitational, Inc. | ||
* | ||
* This program is free software: you can redistribute it and/or modify | ||
* it under the terms of the GNU Affero General Public License as published by | ||
* the Free Software Foundation, either version 3 of the License, or | ||
* (at your option) any later version. | ||
* | ||
* This program is distributed in the hope that it will be useful, | ||
* but WITHOUT ANY WARRANTY; without even the implied warranty of | ||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | ||
* GNU Affero General Public License for more details. | ||
* | ||
* You should have received a copy of the GNU Affero General Public License | ||
* along with this program. If not, see <http://www.gnu.org/licenses/>. | ||
*/ | ||
|
||
package log | ||
|
||
import ( | ||
"context" | ||
"errors" | ||
"io/fs" | ||
"log/slog" | ||
"os" | ||
"path/filepath" | ||
"sync" | ||
|
||
"github.com/fsnotify/fsnotify" | ||
"github.com/gravitational/trace" | ||
) | ||
|
||
var ( | ||
// ErrFileSharedWriterClosed is returned when file shared writer is already closed. | ||
ErrFileSharedWriterClosed = errors.New("file shared writer is closed") | ||
) | ||
|
||
// FileSharedWriter is similar to SharedWriter except that it requires a `os.File` instead of a `io.Writer`. | ||
// This is to allow the File reopen required by logrotate and similar tools. | ||
type FileSharedWriter struct { | ||
logFileName string | ||
fileFlag int | ||
fileMode fs.FileMode | ||
file *os.File | ||
watcher *fsnotify.Watcher | ||
closed bool | ||
|
||
lock sync.Mutex | ||
} | ||
|
||
// NewFileSharedWriter wraps the provided [os.File] in a writer that is thread safe, | ||
// with ability to enable filesystem notification watch and reopen file on specific events. | ||
func NewFileSharedWriter(logFileName string, flag int, mode fs.FileMode) (*FileSharedWriter, error) { | ||
if logFileName == "" { | ||
return nil, trace.BadParameter("log file name is not set") | ||
} | ||
logFile, err := os.OpenFile(logFileName, flag, mode) | ||
if err != nil { | ||
return nil, trace.ConvertSystemError(err) | ||
} | ||
watcher, err := fsnotify.NewWatcher() | ||
if err != nil { | ||
return nil, trace.Wrap(err) | ||
} | ||
|
||
return &FileSharedWriter{ | ||
logFileName: logFileName, | ||
fileFlag: flag, | ||
fileMode: mode, | ||
file: logFile, | ||
watcher: watcher, | ||
}, nil | ||
} | ||
|
||
// Write writes len(b) bytes from b to the File. | ||
func (s *FileSharedWriter) Write(b []byte) (int, error) { | ||
s.lock.Lock() | ||
defer s.lock.Unlock() | ||
if s.closed { | ||
return 0, ErrFileSharedWriterClosed | ||
} | ||
|
||
return s.file.Write(b) | ||
} | ||
|
||
// Reopen closes the file and opens it again using APPEND mode. | ||
func (s *FileSharedWriter) Reopen() error { | ||
// If opening the file is locked we should not acquire a lock and block write. | ||
file, err := os.OpenFile(s.logFileName, s.fileFlag, s.fileMode) | ||
if err != nil { | ||
return trace.Wrap(err) | ||
} | ||
|
||
s.lock.Lock() | ||
if s.closed { | ||
vapopov marked this conversation as resolved.
Show resolved
Hide resolved
|
||
s.lock.Unlock() | ||
_ = file.Close() | ||
return trace.Wrap(ErrFileSharedWriterClosed) | ||
} | ||
oldLogFile := s.file | ||
s.file = file | ||
s.lock.Unlock() | ||
|
||
return trace.Wrap(oldLogFile.Close()) | ||
} | ||
|
||
// RunWatcherReopen runs a filesystem watcher for rename/remove events to reopen the log. | ||
func (s *FileSharedWriter) RunWatcherReopen(ctx context.Context) error { | ||
s.lock.Lock() | ||
defer s.lock.Unlock() | ||
if s.closed { | ||
return trace.Wrap(ErrFileSharedWriterClosed) | ||
} | ||
|
||
return s.runWatcherFunc(ctx, s.Reopen) | ||
} | ||
|
||
// runWatcherFunc spawns goroutine with the watcher loop to consume events of renaming | ||
// or removing the log file to trigger the action function when event appeared. | ||
func (s *FileSharedWriter) runWatcherFunc(ctx context.Context, action func() error) error { | ||
go func() { | ||
for { | ||
select { | ||
case event, ok := <-s.watcher.Events: | ||
if !ok { | ||
return | ||
} | ||
if s.logFileName == event.Name && (event.Has(fsnotify.Rename) || event.Has(fsnotify.Remove)) { | ||
slog.DebugContext(ctx, "Log file was moved/removed", "file", event.Name) | ||
if err := action(); err != nil { | ||
slog.ErrorContext(ctx, "Failed to reopen file", "error", err, "file", event.Name) | ||
continue | ||
} | ||
} | ||
case err, ok := <-s.watcher.Errors: | ||
if !ok { | ||
return | ||
} | ||
slog.ErrorContext(ctx, "Error received on logger watcher", "error", err) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Do we need to do anything else if we get an error, other than log it? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. good question, possible errors, I've checked // reads events from inotify
n, err := w.inotifyFile.Read(buf[:]) if n < unix.SizeofInotifyEvent // Buffer for a maximum of 4096 raw events
// fsnotify: queue or buffer overflow
if mask&unix.IN_Q_OVERFLOW != 0 if watch != nil && mask&unix.IN_MOVE_SELF == unix.IN_MOVE_SELF
// and then, when we remove watching path if somehow its already removed we trigger the error sending the error to the channel doesn't stop internal loop of reading events, another interesting case is
so if we move the log directory somewhere else, we will not able to re-open file since there are no directory, so possible solution it is re-create it (recursively since it might be any depth), or raise an error message in log file which should still writes to old file There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
We don't create the directories when we open the log file the first time, right? I'm not sure what the behavior should be in that case, but it seems really awkward to watch for all the directories in the supposed path just to figure out when the file can be reopened, so maybe that's just not our problem and if the directory is moved or deleted we just give up on watching? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. When we start watching the directory, we actually get event if this specific directory renamed as well, agree with you about creation since we don't know parameters to create it, just was suggestion. If directory is moved There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Can we pick the initial fsmode of the parent directory and create it using the same value? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @tigrato it might unpredictable behavior, since we keep writing in the moved file with directory (reopen fails) we should be good just to raise an error |
||
} | ||
} | ||
}() | ||
|
||
logDirParent := filepath.Dir(s.logFileName) | ||
if err := s.watcher.Add(logDirParent); err != nil { | ||
return trace.Wrap(err) | ||
} | ||
|
||
return nil | ||
} | ||
|
||
// Close stops the internal watcher and close the log file. | ||
func (s *FileSharedWriter) Close() error { | ||
s.lock.Lock() | ||
defer s.lock.Unlock() | ||
|
||
vapopov marked this conversation as resolved.
Show resolved
Hide resolved
|
||
if s.closed { | ||
return trace.Wrap(ErrFileSharedWriterClosed) | ||
} | ||
s.closed = true | ||
|
||
return trace.NewAggregate(s.watcher.Close(), s.file.Close()) | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Should we have a separate lock for the write synchronization and for the rest of the FileSharedWriter machinery?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Not sure if we really need it, because we can't write when in goroutine reopen is triggered, same for closing the file. Only
RunWatcherReopen
kind of independent of the writing process, but we call it before file shared writer is assigned to the logger (it might be even moved to constructor)