From 799b9b69ba08b898e13026b7ecab9f9f58580a82 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Tue, 8 Dec 2020 14:49:41 +0000 Subject: merge develop --- internal/io/logger/logger.go | 2 +- internal/io/signal/signal.go | 29 +++++++++++++++++++++-------- 2 files changed, 22 insertions(+), 9 deletions(-) (limited to 'internal/io') diff --git a/internal/io/logger/logger.go b/internal/io/logger/logger.go index d059cbb..b7db0a7 100644 --- a/internal/io/logger/logger.go +++ b/internal/io/logger/logger.go @@ -224,7 +224,7 @@ func log(what string, severity string, args []interface{}) string { return "" } - messages := []string{severity} + messages := []string{} for _, arg := range args { switch v := arg.(type) { diff --git a/internal/io/signal/signal.go b/internal/io/signal/signal.go index bca7e6e..500c530 100644 --- a/internal/io/signal/signal.go +++ b/internal/io/signal/signal.go @@ -5,24 +5,37 @@ import ( "os" gosignal "os/signal" "syscall" + "time" + + "github.com/mimecast/dtail/internal/config" ) -// StatsCh returns a channel for "please print stats" signalling. -func StatsCh(ctx context.Context) <-chan struct{} { - sigCh := make(chan os.Signal) - gosignal.Notify(sigCh, syscall.SIGINFO, syscall.SIGUSR1) +// InterruptCh returns a channel for "please print stats" signalling. +func InterruptCh(ctx context.Context) <-chan string { + sigIntCh := make(chan os.Signal) + gosignal.Notify(sigIntCh, os.Interrupt) + + sigOtherCh := make(chan os.Signal) + gosignal.Notify(sigOtherCh, syscall.SIGHUP, syscall.SIGTERM, syscall.SIGQUIT) - statsCh := make(chan struct{}) + statsCh := make(chan string) go func() { for { select { - case <-sigCh: + case <-sigIntCh: select { - case statsCh <- struct{}{}: + case statsCh <- "Hint: Hit Ctrl+C again to exit": + select { + case <-sigIntCh: + os.Exit(0) + case <-time.After(time.Second * time.Duration(config.InterruptTimeoutS)): + } default: - // Stats currently already printed. + // Stats already printed. } + case <-sigOtherCh: + os.Exit(0) case <-ctx.Done(): return } -- cgit v1.2.3 From ab676c2b484225ed22765b23d8f0545088ecd610 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 26 Dec 2020 10:48:51 +0000 Subject: code cleanup and minor refactorings --- internal/io/line/line.go | 6 +- internal/io/prompt/prompt.go | 6 +- internal/io/run/run.go | 150 ------------------------------------------- 3 files changed, 8 insertions(+), 154 deletions(-) delete mode 100644 internal/io/run/run.go (limited to 'internal/io') diff --git a/internal/io/line/line.go b/internal/io/line/line.go index 9db93c0..715be34 100644 --- a/internal/io/line/line.go +++ b/internal/io/line/line.go @@ -15,7 +15,11 @@ type Line struct { // lines if that happens but it will signal to the client how // many log lines in % could be transmitted to the client. TransmittedPerc int - SourceID string + // Contains the unique identifier of the source log file. + // It could be the name of the log or it could be one of the parent + // directories in case multiple log files with the same basename are + // followed. + SourceID string } // Return a human readable representation of the followed line. diff --git a/internal/io/prompt/prompt.go b/internal/io/prompt/prompt.go index a438d33..36ebdb5 100644 --- a/internal/io/prompt/prompt.go +++ b/internal/io/prompt/prompt.go @@ -3,9 +3,10 @@ package prompt import ( "bufio" "fmt" - "github.com/mimecast/dtail/internal/io/logger" "os" "strings" + + "github.com/mimecast/dtail/internal/io/logger" ) // Answer is a user input of a prompt question. @@ -18,8 +19,7 @@ type Answer struct { Callback func() // Runs after Callback and after logging resumes EndCallback func() - - AskAgain bool + AskAgain bool } // Prompt used for interactive user input. diff --git a/internal/io/run/run.go b/internal/io/run/run.go deleted file mode 100644 index 2bb3756..0000000 --- a/internal/io/run/run.go +++ /dev/null @@ -1,150 +0,0 @@ -package run - -import ( - "bufio" - "context" - "io" - "os/exec" - "sync" - "syscall" - "time" - - "github.com/mimecast/dtail/internal/io/line" - "github.com/mimecast/dtail/internal/io/logger" -) - -// Run is for execute a command. -type Run struct { - command string - args []string - cmd *exec.Cmd - pgroupKilled chan struct{} -} - -// New returns a new command runner. -func New(command string, args []string) Run { - return Run{ - command: command, - args: args, - pgroupKilled: make(chan struct{}), - } -} - -// StartBackground starts running the command in background. -func (r Run) StartBackground(ctx context.Context, wg *sync.WaitGroup, ec chan<- int, lines chan<- line.Line) (pid int, err error) { - pid = -1 - - if len(r.args) > 0 { - logger.Debug(r.command, r.args, " ") - r.cmd = exec.CommandContext(ctx, r.command, r.args...) - } else { - logger.Debug(r.command) - r.cmd = exec.CommandContext(ctx, r.command) - } - - // Create a new process group, so that kill() will only kill this command + pgroup. - r.cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} - - stdoutPipe, myErr := r.cmd.StdoutPipe() - if err != nil { - wg.Done() - err = myErr - return - } - - stderrPipe, myErr := r.cmd.StderrPipe() - if myErr != nil { - wg.Done() - err = myErr - return - } - - if myErr := r.cmd.Start(); err != nil { - wg.Done() - err = myErr - return - } - - if r.cmd.Process != nil { - pid = r.cmd.Process.Pid - } - - commandExited := make(chan struct{}) - - var pipeWg sync.WaitGroup - pipeWg.Add(2) - - go r.killPgroup(ctx, commandExited, pid) - go r.pipeToLines(commandExited, &pipeWg, pid, stdoutPipe, "STDOUT", lines) - go r.pipeToLines(commandExited, &pipeWg, pid, stderrPipe, "STDERR", lines) - - go func() { - exitCode := 255 - if waitErr := r.cmd.Wait(); waitErr != nil { - if exitError, ok := waitErr.(*exec.ExitError); ok { - exitCode = exitError.ExitCode() - } - } - ec <- exitCode - - // Tell pipes we are done - close(commandExited) - // Wait for process group to be killed - <-r.pgroupKilled - // Wait for the pipes to flush the contents - pipeWg.Wait() - // Now the job is truly done - wg.Done() - }() - - return -} - -func (r Run) pipeToLines(commandExited chan struct{}, wg *sync.WaitGroup, pid int, reader io.Reader, what string, lines chan<- line.Line) { - defer wg.Done() - bufReader := bufio.NewReader(reader) - - for { - time.Sleep(time.Millisecond * 10) - lineStr, err := bufReader.ReadString('\n') - - if err != nil { - select { - case <-commandExited: - return - default: - } - continue - } - - newLine := line.Line{ - Content: []byte(lineStr), - Count: uint64(pid), - TransmittedPerc: 100, - SourceID: what, - } - - select { - case lines <- newLine: - case <-commandExited: - return - } - } -} - -func (r Run) killPgroup(ctx context.Context, commandExited chan struct{}, pid int) { - if pid == -1 { - close(r.pgroupKilled) - return - } - - if pgid, err := syscall.Getpgid(pid); err == nil { - // Kill process group when done - select { - case <-ctx.Done(): - case <-commandExited: - } - syscall.Kill(-pgid, syscall.SIGKILL) - close(r.pgroupKilled) - } -} -- cgit v1.2.3 From f9415259bc5d74a89801be75587da87dd37518af Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 26 Dec 2020 14:28:31 +0000 Subject: initial quiet switch --- internal/io/logger/logger.go | 22 ++++++++++++++-------- internal/io/logger/modes.go | 1 + 2 files changed, 15 insertions(+), 8 deletions(-) (limited to 'internal/io') diff --git a/internal/io/logger/logger.go b/internal/io/logger/logger.go index b7db0a7..7674dd1 100644 --- a/internal/io/logger/logger.go +++ b/internal/io/logger/logger.go @@ -72,12 +72,15 @@ type buf struct { func Start(ctx context.Context, mode Modes) { Mode = mode - if Mode.Nothing { + switch { + case Mode.Nothing: return - } - - if Mode.Trace { + case Mode.Quiet: + Mode.Trace = false + Mode.Debug = false + case Mode.Trace: Mode.Debug = true + default: } strategy := logStrategy() @@ -87,7 +90,7 @@ func Start(ctx context.Context, mode Modes) { case DailyStrategy: _, err := os.Stat(config.Common.LogDir) Mode.logToFile = !os.IsNotExist(err) - Mode.logToStdout = !Mode.Server || Mode.Debug || Mode.Trace + Mode.logToStdout = !Mode.Server || Mode.Debug || Mode.Trace || Mode.Quiet case StdoutStrategy: fallthrough default: @@ -131,11 +134,14 @@ func Info(args ...interface{}) string { // Warn message logging. func Warn(args ...interface{}) string { - if Mode.Server { - return log(serverStr, warnStr, args) + if !Mode.Quiet { + if Mode.Server { + return log(serverStr, warnStr, args) + } + return log(clientStr, warnStr, args) } - return log(clientStr, warnStr, args) + return "" } // Error message logging. diff --git a/internal/io/logger/modes.go b/internal/io/logger/modes.go index 47dadfe..8864179 100644 --- a/internal/io/logger/modes.go +++ b/internal/io/logger/modes.go @@ -6,6 +6,7 @@ type Modes struct { Trace bool Debug bool Nothing bool + Quiet bool logToStdout bool logToFile bool } -- cgit v1.2.3 From eca9c65b7c9e33cba8cc1ea5afe016bfc59f8918 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 26 Dec 2020 14:34:32 +0000 Subject: correctly format server messages --- internal/io/logger/logger.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'internal/io') diff --git a/internal/io/logger/logger.go b/internal/io/logger/logger.go index 7674dd1..4254eef 100644 --- a/internal/io/logger/logger.go +++ b/internal/io/logger/logger.go @@ -248,7 +248,7 @@ func log(what string, severity string, args []interface{}) string { message := strings.Join(messages, "|") write(what, severity, message) - return message + return fmt.Sprintf("%s|%s", severity, message) } // Raw message logging. -- cgit v1.2.3 From 2c7bdd09e8b7c58d98d631e32a24e4bd34d5bec9 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sun, 27 Dec 2020 16:42:03 +0000 Subject: make lint happy --- internal/io/fs/permissions/permission_linux.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'internal/io') diff --git a/internal/io/fs/permissions/permission_linux.go b/internal/io/fs/permissions/permission_linux.go index feae729..bbc039b 100644 --- a/internal/io/fs/permissions/permission_linux.go +++ b/internal/io/fs/permissions/permission_linux.go @@ -11,7 +11,7 @@ import ( "unsafe" ) -// To check whether user has Linux file system permissions to read a given file. +// ToRead checks whether user has Linux file system permissions to read a given file. func ToRead(user, filePath string) (bool, error) { cUser := C.CString(user) cFilePath := C.CString(filePath) -- cgit v1.2.3 From 0099a7ab9e1d28300c69c3b50b4ebe1cde9a8cbc Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Tue, 29 Dec 2020 08:34:04 +0000 Subject: Make Linux ACL support optional, as it requires CGo and makes the binary less portable --- internal/io/fs/permissions/permission.go | 2 +- internal/io/fs/permissions/permission_linux.c | 395 --------------------- internal/io/fs/permissions/permission_linux.go | 33 -- internal/io/fs/permissions/permission_linux.h | 60 ---- internal/io/fs/permissions/permission_linuxacl.c | 397 ++++++++++++++++++++++ internal/io/fs/permissions/permission_linuxacl.go | 35 ++ internal/io/fs/permissions/permission_linuxacl.h | 62 ++++ internal/io/fs/permissions/permission_test.go | 2 +- 8 files changed, 496 insertions(+), 490 deletions(-) delete mode 100644 internal/io/fs/permissions/permission_linux.c delete mode 100644 internal/io/fs/permissions/permission_linux.go delete mode 100644 internal/io/fs/permissions/permission_linux.h create mode 100644 internal/io/fs/permissions/permission_linuxacl.c create mode 100644 internal/io/fs/permissions/permission_linuxacl.go create mode 100644 internal/io/fs/permissions/permission_linuxacl.h (limited to 'internal/io') diff --git a/internal/io/fs/permissions/permission.go b/internal/io/fs/permissions/permission.go index 0ed4f17..cc5dd9b 100644 --- a/internal/io/fs/permissions/permission.go +++ b/internal/io/fs/permissions/permission.go @@ -1,4 +1,4 @@ -// +build !linux +// +build !linuxacl package permissions diff --git a/internal/io/fs/permissions/permission_linux.c b/internal/io/fs/permissions/permission_linux.c deleted file mode 100644 index cd10525..0000000 --- a/internal/io/fs/permissions/permission_linux.c +++ /dev/null @@ -1,395 +0,0 @@ -#include "permission_linux.h" - -#ifdef DEBUG -void debug_print_checker(struct permission_checker *pc) { - fprintf(stderr, "DEBUG: user_name:%s (%d)\n", - pc->user_name, pc->uid); - - fprintf(stderr, "DEBUG: ngids:%d\n", pc->ngids); - int j; - for (j = 0; j < pc->ngids; j++) { - fprintf(stderr, "DEBUG: %d", pc->gids[j]); - struct group *gr = getgrgid(pc->gids[j]); - if (gr != NULL) - fprintf(stderr, " (%s)", gr->gr_name); - fprintf(stderr, "\n"); - } - - fprintf(stderr, "DEBUG: file_path:%s (%d:%d)\n", - pc->file_path, pc->file_stat.st_uid, pc->file_stat.st_gid); -} -#endif // DEBUG - -int stat_file(struct permission_checker *pc) { - if (stat(pc->file_path, &pc->file_stat) != 0) - return -1; - -#ifdef DEBUG - fprintf(stderr, "DEBUG: File'%s' is owned by '%d:%d'\n", - pc->file_path, pc->file_stat.st_uid, pc->file_stat.st_gid); -#endif - - return 0; -} - -int get_user_uid(struct permission_checker *pc) { - struct passwd *result = NULL; - - size_t bufsize = sysconf(_SC_GETPW_R_SIZE_MAX); - if (bufsize == -1) - bufsize = 16384; - - char *buf = malloc(bufsize); - if (buf == NULL) { -#ifdef DEBUG - fprintf(stderr, "DEBUG: Unabel to allocate bufer while retrieving user '%s'\n", pc->user_name); -#endif - return -1; - } - - int rc = getpwnam_r(pc->user_name, &pc->pw, buf, bufsize, &result); - - if (result == NULL) { -#ifdef DEBUG - if (rc == 0) { - fprintf(stderr, "DEBUG: No user '%s' found\n", pc->user_name); - } else { - fprintf(stderr, "DEBUG: Unknown error while retrieving user '%s'\n", pc->user_name); - } -#endif - - free(buf); - return -1; - } - - pc->uid = pc->pw.pw_uid; - - free(buf); - return 0; -} - -int get_user_groups(struct permission_checker *pc) { - // First assume we are in 10 groups max - pc->ngids = 10; - pc->gids = malloc(pc->ngids * sizeof(gid_t)); - - if (pc->gids == NULL) { -#ifdef DEBUG - fprintf(stderr, "DEBUG: Unable to allocate space for gids."); -#endif - return -1; - } - - // Try so many times to load group list until it fits into group array. - while (getgrouplist(pc->user_name, pc->pw.pw_gid, pc->gids, &pc->ngids) == -1) { - // Too many groups, enlarge group array and try again - int newngids = pc->ngids + 100; - size_t newsize = newngids * sizeof(gid_t); - - if (SIZE_MAX / newngids < sizeof(gid_t)) { - // Overflow -#ifdef DEBUG - fprintf(stderr, "DEBUG: Overflow detected."); -#endif - return -1; - } - - gid_t *newgids = realloc(pc->gids, newsize); - if (newgids == NULL) { -#ifdef DEBUG - fprintf(stderr, "DEBUG: Unable to allocate space for gids."); -#endif - free(pc->gids); - return -1; - } - - pc->gids = newgids; - pc->ngids = newngids; - } - - return 0; -} - -int is_member_of_group(struct permission_checker *pc, gid_t gid) { - int j; - for (j = 0; j < pc->ngids; j++) - if (pc->gids[j] == gid) - return 1; - return 0; -} - -int check_acl_uid_matches(uid_t uid, acl_entry_t entry) { - int ret = -1; - uid_t *acl_uid = acl_get_qualifier(entry); - if (acl_uid == NULL) { -#ifdef DEBUG - fprintf(stderr, "DEBUG: Unable to retrieve user uid from ACL entry"); -#endif - return -1; - } - - ret = *acl_uid == uid ? 0 : -1; -#ifdef DEBUG - fprintf(stderr, "DEBUG: ACL user match?: %d <=> %d: %d\n", *acl_uid, uid, ret); -#endif - acl_free(acl_uid); - return ret; -} - -int check_acl_gid_matches(gid_t *gids, int ngids, acl_entry_t entry) { - int ret = -1; - gid_t *acl_gid = acl_get_qualifier(entry); - if (acl_gid == NULL) { -#ifdef DEBUG - fprintf(stderr, "DEBUG: Unable to retrieve user uid from ACL entry"); -#endif - return -1; - } - - int j; - for (j = 0; j < ngids; j++) { - if (*acl_gid == gids[j]) { -#ifdef DEBUG - fprintf(stderr, "DEBUG: User is in group %d", *acl_gid); -#endif - ret = 0; - break; - } - } - -#ifdef DEBUG - fprintf(stderr, "DEBUG: ACL group match?: %d <=> ...: %d\n", *acl_gid, ret); -#endif - acl_free(acl_gid); - return ret; -} - -int check_acl(struct permission_checker *pc, const int flag) { - // By default user has no read perm. - int has_read_perm = 0; - - // By default mask tells that there are read perm. However in order to have - // read permissions both, has_read_perm and mask_allows_read_access must be 1! - int mask_allows_read_access = 1; - - acl_type_t type = ACL_TYPE_ACCESS; - acl_t acl = acl_get_file(pc->file_path, type); - - if (acl == NULL) - // Unable to retrieve ACL. - return -1; - - // Walk through each entry of this ACL. - int id; - for (id = ACL_FIRST_ENTRY; ; id = ACL_NEXT_ENTRY) { - acl_entry_t entry; - if (acl_get_entry(acl, id, &entry) != 1) - // No more ACL entries. - break; - - acl_tag_t tag; - if (acl_get_tag_type(entry, &tag) == -1) - // Unable to retrieve ACL tag. - return -1; - - switch (tag) { - case ACL_USER_OBJ: - if (flag == GROUP_CHECK) - continue; -#ifdef DEBUG - fprintf(stderr, "DEBUG: ACL_USER_OBJ\n"); -#endif - // Ignore this ACL entry if user is not owner of file. - if (pc->uid != pc->file_stat.st_uid) - continue; - break; - case ACL_USER: - if (flag == GROUP_CHECK) - continue; -#ifdef DEBUG - fprintf(stderr, "DEBUG: ACL_USER\n"); -#endif - // Ignore this ACL entry if uid does not match. - if (check_acl_uid_matches(pc->uid, entry) != 0) - continue; - break; - case ACL_GROUP_OBJ: - if (flag == USER_CHECK) - continue; -#ifdef DEBUG - fprintf(stderr, "DEBUG: ACL_GROUP_OBJ\n"); -#endif - // Ignore ACL entry if user is not in group of file. - if (!is_member_of_group(pc, pc->file_stat.st_gid)) - continue; - break; - case ACL_GROUP: - if (flag == USER_CHECK) - continue; -#ifdef DEBUG - fprintf(stderr, "DEBUG: ACL_GROUP\n"); -#endif - // Ignore ACL entry if user is not in group of entry. - if (check_acl_gid_matches(pc->gids, pc->ngids, entry) != 0) - continue; - break; - case ACL_OTHER: - if (flag == GROUP_CHECK) - continue; -#ifdef DEBUG - fprintf(stderr, "DEBUG: ACL_OTHER\n"); -#endif - break; - case ACL_MASK: -#ifdef DEBUG - fprintf(stderr, "DEBUG: ACL_MASK\n"); -#endif - break; - default: -#ifdef DEBUG - fprintf(stderr, "DEBUG: Unknown ACL tag\n"); -#endif - return -1; - } - -#ifdef DEBUG - fprintf(stderr, "DEBUG: Retrieving permset\n"); -#endif - acl_permset_t permset; - int permission; - if (acl_get_permset(entry, &permset) == -1) - // Unable to retrieve permset. - return -1; - - if ((permission = acl_get_perm(permset, ACL_READ)) == -1) - // Unable to retrieve permset value. - return -1; - - if (permission == 1 && tag != ACL_MASK) { -#ifdef DEBUG - fprintf(stderr, "DEBUG: ACL says user has permission to read file.\n"); -#endif - has_read_perm = 1; - } else if (permission == 0 && tag == ACL_MASK) { - // Mask says that there are no permissions to read. - mask_allows_read_access = 0; -#ifdef DEBUG - fprintf(stderr, "DEBUG: ACL mask says no permission to read file.\n"); -#endif - } - } - - if (has_read_perm && mask_allows_read_access) { -#ifdef DEBUG - fprintf(stderr, "DEBUG: ACL end result: User has permission to read file.\n"); -#endif - return 1; - } - -#ifdef DEBUG - fprintf(stderr, "DEBUG: ACL end result: User has no permission to read file.\n"); -#endif - return 0; -} - -int check_traditional(struct permission_checker *pc, const int flag) { - mode_t mode = pc->file_stat.st_mode; - uid_t uid = pc->file_stat.st_uid; - gid_t gid = pc->file_stat.st_gid; - - if (flag == USER_CHECK && (mode & S_IROTH)) { -#ifdef DEBUG - fprintf(stderr, "DEBUG: Others can read file '%s'\n", - pc->file_path); -#endif - return 1; - - } else if (flag == USER_CHECK && (mode & S_IRUSR) && uid == pc->uid) { -#ifdef DEBUG - fprintf(stderr, "DEBUG: User '%s' can read file '%s'\n", - pc->user_name, pc->file_path); -#endif - return 1; - - } else if (flag == GROUP_CHECK && (mode & S_IRGRP) && is_member_of_group(pc, gid)) { -#ifdef DEBUG - fprintf(stderr, "DEBUG: User's '%s' group can read file '%s'\n", - pc->user_name, pc->file_path); -#endif - return 1; - } - - return 0; -} - -int permission_to_read(char* user_name, char *file_path) { - int rc = -1; - -#ifdef DEBUG - fprintf(stderr, "DEBUG: User check '%s' for file '%s'\n", user_name, file_path); -#endif - struct permission_checker pc = { - .user_name = user_name, - .gids = NULL, - .ngids = 0, - .file_path = file_path, - }; - - // Gather user's UID. - if ((rc = get_user_uid(&pc)) == -1) - // Could not retrieve UID. - goto cleanup; - - // Gather file owner (user and group). - if ((rc = stat_file(&pc)) == -1) - // Could not stat file. - goto cleanup; - - // Check whether there is an ACL entry which would allow the user - // to read the file. Don't check for any groups yet. The issue with - // groups is that it can be very slow to retrieve the list of groups - // of a specific user when done via a remote LDAP server! - if ((rc = check_acl(&pc, USER_CHECK)) == 1) - // Yes, has permissions. - goto cleanup; - - // Check whether ACLs of file could be retrieved. - if (rc == -1) { - if (errno != ENOTSUP) - // Unknown error. - goto cleanup; - - // File system does not support ACLs. - // Fallback to traditional permissions. - if ((rc = check_traditional(&pc, USER_CHECK)) == 1) - // Yes, has traditional permissions. - goto cleanup; - - if ((rc = get_user_groups(&pc)) == -1) - // Can not retrieve user's groups. - goto cleanup; - - rc = check_traditional(&pc, GROUP_CHECK); - goto cleanup; - } - - if ((rc = get_user_groups(&pc)) == -1) - // Can not retrieve use'r groups. - goto cleanup; - - // Check whether there is an ACL entry which would allow any of the - // user's groups to read the file. - rc = check_acl(&pc, GROUP_CHECK); - -cleanup: -#ifdef DEBUG - debug_print_checker(&pc); -#endif - - if (pc.ngids) - free(pc.gids); - - return rc; -} - -// vim: set tabstop=8 softtabstop=0 expandtab shiftwidth=4 smarttab diff --git a/internal/io/fs/permissions/permission_linux.go b/internal/io/fs/permissions/permission_linux.go deleted file mode 100644 index bbc039b..0000000 --- a/internal/io/fs/permissions/permission_linux.go +++ /dev/null @@ -1,33 +0,0 @@ -package permissions - -/* -#include "permission_linux.h" -#cgo LDFLAGS: -L. -lacl -*/ -import "C" - -import ( - "errors" - "unsafe" -) - -// ToRead checks whether user has Linux file system permissions to read a given file. -func ToRead(user, filePath string) (bool, error) { - cUser := C.CString(user) - cFilePath := C.CString(filePath) - - defer C.free(unsafe.Pointer(cUser)) - defer C.free(unsafe.Pointer(cFilePath)) - - cOk, err := C.permission_to_read(cUser, cFilePath) - if cOk == 1 { - return true, nil - } - - if err != nil { - // err contains errno message - return false, err - } - - return false, errors.New("User without permission to read file") -} diff --git a/internal/io/fs/permissions/permission_linux.h b/internal/io/fs/permissions/permission_linux.h deleted file mode 100644 index a2c266e..0000000 --- a/internal/io/fs/permissions/permission_linux.h +++ /dev/null @@ -1,60 +0,0 @@ -#ifndef PERMISSION_LINUX_H -#define PERMISSION_LINUX_H - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -//#define DEBUG -#define USER_CHECK 0 -#define GROUP_CHECK 1 - -struct permission_checker { - char *user_name; - uid_t uid; - gid_t *gids; - int ngids; - char *file_path; - struct stat file_stat; - struct passwd pw; -}; - - -#ifdef DEBUG -// Print out permission_checker struct. -void debug_print_checker(struct permission_checker *pc); -#endif - -// Stat a given file to retrieve traditional UNIX permissions. -int stat_file(struct permission_checker *pc); - -// Retrieve UID of user. -int get_user_uid(struct permission_checker *pc); - -// Retrieve all groups of the user. -int get_user_groups(struct permission_checker *pc); - -// Check whether user is member of a group or not. -int is_member_of_group(struct permission_checker *pc, gid_t gid); - -// Check whether user can read file according Linux ACLs. -// As flag use either USER_CHECK or GROUP_CHECK. -int check_acl(struct permission_checker *pc, const int flag); - -// Check whether user has permissions to read file according traditional -// UNIX permissions. As flag use either USER_CHECK or GROUP_CHECK. -int check_traditional(struct permission_checker *pc, const int flag); - -// Returns 1 if user has permission to read file. -// Returns <0 on error and returns 0 if no permissions. -int permission_to_read(char* user, char *file_path); - -#endif // PERMISSION_LINUX_H diff --git a/internal/io/fs/permissions/permission_linuxacl.c b/internal/io/fs/permissions/permission_linuxacl.c new file mode 100644 index 0000000..86b1185 --- /dev/null +++ b/internal/io/fs/permissions/permission_linuxacl.c @@ -0,0 +1,397 @@ +// +build linuxacl + +#include "permission_linuxacl.h" + +#ifdef DEBUG +void debug_print_checker(struct permission_checker *pc) { + fprintf(stderr, "DEBUG: user_name:%s (%d)\n", + pc->user_name, pc->uid); + + fprintf(stderr, "DEBUG: ngids:%d\n", pc->ngids); + int j; + for (j = 0; j < pc->ngids; j++) { + fprintf(stderr, "DEBUG: %d", pc->gids[j]); + struct group *gr = getgrgid(pc->gids[j]); + if (gr != NULL) + fprintf(stderr, " (%s)", gr->gr_name); + fprintf(stderr, "\n"); + } + + fprintf(stderr, "DEBUG: file_path:%s (%d:%d)\n", + pc->file_path, pc->file_stat.st_uid, pc->file_stat.st_gid); +} +#endif // DEBUG + +int stat_file(struct permission_checker *pc) { + if (stat(pc->file_path, &pc->file_stat) != 0) + return -1; + +#ifdef DEBUG + fprintf(stderr, "DEBUG: File'%s' is owned by '%d:%d'\n", + pc->file_path, pc->file_stat.st_uid, pc->file_stat.st_gid); +#endif + + return 0; +} + +int get_user_uid(struct permission_checker *pc) { + struct passwd *result = NULL; + + size_t bufsize = sysconf(_SC_GETPW_R_SIZE_MAX); + if (bufsize == -1) + bufsize = 16384; + + char *buf = malloc(bufsize); + if (buf == NULL) { +#ifdef DEBUG + fprintf(stderr, "DEBUG: Unabel to allocate bufer while retrieving user '%s'\n", pc->user_name); +#endif + return -1; + } + + int rc = getpwnam_r(pc->user_name, &pc->pw, buf, bufsize, &result); + + if (result == NULL) { +#ifdef DEBUG + if (rc == 0) { + fprintf(stderr, "DEBUG: No user '%s' found\n", pc->user_name); + } else { + fprintf(stderr, "DEBUG: Unknown error while retrieving user '%s'\n", pc->user_name); + } +#endif + + free(buf); + return -1; + } + + pc->uid = pc->pw.pw_uid; + + free(buf); + return 0; +} + +int get_user_groups(struct permission_checker *pc) { + // First assume we are in 10 groups max + pc->ngids = 10; + pc->gids = malloc(pc->ngids * sizeof(gid_t)); + + if (pc->gids == NULL) { +#ifdef DEBUG + fprintf(stderr, "DEBUG: Unable to allocate space for gids."); +#endif + return -1; + } + + // Try so many times to load group list until it fits into group array. + while (getgrouplist(pc->user_name, pc->pw.pw_gid, pc->gids, &pc->ngids) == -1) { + // Too many groups, enlarge group array and try again + int newngids = pc->ngids + 100; + size_t newsize = newngids * sizeof(gid_t); + + if (SIZE_MAX / newngids < sizeof(gid_t)) { + // Overflow +#ifdef DEBUG + fprintf(stderr, "DEBUG: Overflow detected."); +#endif + return -1; + } + + gid_t *newgids = realloc(pc->gids, newsize); + if (newgids == NULL) { +#ifdef DEBUG + fprintf(stderr, "DEBUG: Unable to allocate space for gids."); +#endif + free(pc->gids); + return -1; + } + + pc->gids = newgids; + pc->ngids = newngids; + } + + return 0; +} + +int is_member_of_group(struct permission_checker *pc, gid_t gid) { + int j; + for (j = 0; j < pc->ngids; j++) + if (pc->gids[j] == gid) + return 1; + return 0; +} + +int check_acl_uid_matches(uid_t uid, acl_entry_t entry) { + int ret = -1; + uid_t *acl_uid = acl_get_qualifier(entry); + if (acl_uid == NULL) { +#ifdef DEBUG + fprintf(stderr, "DEBUG: Unable to retrieve user uid from ACL entry"); +#endif + return -1; + } + + ret = *acl_uid == uid ? 0 : -1; +#ifdef DEBUG + fprintf(stderr, "DEBUG: ACL user match?: %d <=> %d: %d\n", *acl_uid, uid, ret); +#endif + acl_free(acl_uid); + return ret; +} + +int check_acl_gid_matches(gid_t *gids, int ngids, acl_entry_t entry) { + int ret = -1; + gid_t *acl_gid = acl_get_qualifier(entry); + if (acl_gid == NULL) { +#ifdef DEBUG + fprintf(stderr, "DEBUG: Unable to retrieve user uid from ACL entry"); +#endif + return -1; + } + + int j; + for (j = 0; j < ngids; j++) { + if (*acl_gid == gids[j]) { +#ifdef DEBUG + fprintf(stderr, "DEBUG: User is in group %d", *acl_gid); +#endif + ret = 0; + break; + } + } + +#ifdef DEBUG + fprintf(stderr, "DEBUG: ACL group match?: %d <=> ...: %d\n", *acl_gid, ret); +#endif + acl_free(acl_gid); + return ret; +} + +int check_acl(struct permission_checker *pc, const int flag) { + // By default user has no read perm. + int has_read_perm = 0; + + // By default mask tells that there are read perm. However in order to have + // read permissions both, has_read_perm and mask_allows_read_access must be 1! + int mask_allows_read_access = 1; + + acl_type_t type = ACL_TYPE_ACCESS; + acl_t acl = acl_get_file(pc->file_path, type); + + if (acl == NULL) + // Unable to retrieve ACL. + return -1; + + // Walk through each entry of this ACL. + int id; + for (id = ACL_FIRST_ENTRY; ; id = ACL_NEXT_ENTRY) { + acl_entry_t entry; + if (acl_get_entry(acl, id, &entry) != 1) + // No more ACL entries. + break; + + acl_tag_t tag; + if (acl_get_tag_type(entry, &tag) == -1) + // Unable to retrieve ACL tag. + return -1; + + switch (tag) { + case ACL_USER_OBJ: + if (flag == GROUP_CHECK) + continue; +#ifdef DEBUG + fprintf(stderr, "DEBUG: ACL_USER_OBJ\n"); +#endif + // Ignore this ACL entry if user is not owner of file. + if (pc->uid != pc->file_stat.st_uid) + continue; + break; + case ACL_USER: + if (flag == GROUP_CHECK) + continue; +#ifdef DEBUG + fprintf(stderr, "DEBUG: ACL_USER\n"); +#endif + // Ignore this ACL entry if uid does not match. + if (check_acl_uid_matches(pc->uid, entry) != 0) + continue; + break; + case ACL_GROUP_OBJ: + if (flag == USER_CHECK) + continue; +#ifdef DEBUG + fprintf(stderr, "DEBUG: ACL_GROUP_OBJ\n"); +#endif + // Ignore ACL entry if user is not in group of file. + if (!is_member_of_group(pc, pc->file_stat.st_gid)) + continue; + break; + case ACL_GROUP: + if (flag == USER_CHECK) + continue; +#ifdef DEBUG + fprintf(stderr, "DEBUG: ACL_GROUP\n"); +#endif + // Ignore ACL entry if user is not in group of entry. + if (check_acl_gid_matches(pc->gids, pc->ngids, entry) != 0) + continue; + break; + case ACL_OTHER: + if (flag == GROUP_CHECK) + continue; +#ifdef DEBUG + fprintf(stderr, "DEBUG: ACL_OTHER\n"); +#endif + break; + case ACL_MASK: +#ifdef DEBUG + fprintf(stderr, "DEBUG: ACL_MASK\n"); +#endif + break; + default: +#ifdef DEBUG + fprintf(stderr, "DEBUG: Unknown ACL tag\n"); +#endif + return -1; + } + +#ifdef DEBUG + fprintf(stderr, "DEBUG: Retrieving permset\n"); +#endif + acl_permset_t permset; + int permission; + if (acl_get_permset(entry, &permset) == -1) + // Unable to retrieve permset. + return -1; + + if ((permission = acl_get_perm(permset, ACL_READ)) == -1) + // Unable to retrieve permset value. + return -1; + + if (permission == 1 && tag != ACL_MASK) { +#ifdef DEBUG + fprintf(stderr, "DEBUG: ACL says user has permission to read file.\n"); +#endif + has_read_perm = 1; + } else if (permission == 0 && tag == ACL_MASK) { + // Mask says that there are no permissions to read. + mask_allows_read_access = 0; +#ifdef DEBUG + fprintf(stderr, "DEBUG: ACL mask says no permission to read file.\n"); +#endif + } + } + + if (has_read_perm && mask_allows_read_access) { +#ifdef DEBUG + fprintf(stderr, "DEBUG: ACL end result: User has permission to read file.\n"); +#endif + return 1; + } + +#ifdef DEBUG + fprintf(stderr, "DEBUG: ACL end result: User has no permission to read file.\n"); +#endif + return 0; +} + +int check_traditional(struct permission_checker *pc, const int flag) { + mode_t mode = pc->file_stat.st_mode; + uid_t uid = pc->file_stat.st_uid; + gid_t gid = pc->file_stat.st_gid; + + if (flag == USER_CHECK && (mode & S_IROTH)) { +#ifdef DEBUG + fprintf(stderr, "DEBUG: Others can read file '%s'\n", + pc->file_path); +#endif + return 1; + + } else if (flag == USER_CHECK && (mode & S_IRUSR) && uid == pc->uid) { +#ifdef DEBUG + fprintf(stderr, "DEBUG: User '%s' can read file '%s'\n", + pc->user_name, pc->file_path); +#endif + return 1; + + } else if (flag == GROUP_CHECK && (mode & S_IRGRP) && is_member_of_group(pc, gid)) { +#ifdef DEBUG + fprintf(stderr, "DEBUG: User's '%s' group can read file '%s'\n", + pc->user_name, pc->file_path); +#endif + return 1; + } + + return 0; +} + +int permission_to_read(char* user_name, char *file_path) { + int rc = -1; + +#ifdef DEBUG + fprintf(stderr, "DEBUG: User check '%s' for file '%s'\n", user_name, file_path); +#endif + struct permission_checker pc = { + .user_name = user_name, + .gids = NULL, + .ngids = 0, + .file_path = file_path, + }; + + // Gather user's UID. + if ((rc = get_user_uid(&pc)) == -1) + // Could not retrieve UID. + goto cleanup; + + // Gather file owner (user and group). + if ((rc = stat_file(&pc)) == -1) + // Could not stat file. + goto cleanup; + + // Check whether there is an ACL entry which would allow the user + // to read the file. Don't check for any groups yet. The issue with + // groups is that it can be very slow to retrieve the list of groups + // of a specific user when done via a remote LDAP server! + if ((rc = check_acl(&pc, USER_CHECK)) == 1) + // Yes, has permissions. + goto cleanup; + + // Check whether ACLs of file could be retrieved. + if (rc == -1) { + if (errno != ENOTSUP) + // Unknown error. + goto cleanup; + + // File system does not support ACLs. + // Fallback to traditional permissions. + if ((rc = check_traditional(&pc, USER_CHECK)) == 1) + // Yes, has traditional permissions. + goto cleanup; + + if ((rc = get_user_groups(&pc)) == -1) + // Can not retrieve user's groups. + goto cleanup; + + rc = check_traditional(&pc, GROUP_CHECK); + goto cleanup; + } + + if ((rc = get_user_groups(&pc)) == -1) + // Can not retrieve use'r groups. + goto cleanup; + + // Check whether there is an ACL entry which would allow any of the + // user's groups to read the file. + rc = check_acl(&pc, GROUP_CHECK); + +cleanup: +#ifdef DEBUG + debug_print_checker(&pc); +#endif + + if (pc.ngids) + free(pc.gids); + + return rc; +} + +// vim: set tabstop=8 softtabstop=0 expandtab shiftwidth=4 smarttab diff --git a/internal/io/fs/permissions/permission_linuxacl.go b/internal/io/fs/permissions/permission_linuxacl.go new file mode 100644 index 0000000..7d2d7ca --- /dev/null +++ b/internal/io/fs/permissions/permission_linuxacl.go @@ -0,0 +1,35 @@ +// +build linuxacl + +package permissions + +/* +#include "permission_linuxacl.h" +#cgo LDFLAGS: -L. -lacl +*/ +import "C" + +import ( + "errors" + "unsafe" +) + +// ToRead checks whether user has Linux file system permissions to read a given file. +func ToRead(user, filePath string) (bool, error) { + cUser := C.CString(user) + cFilePath := C.CString(filePath) + + defer C.free(unsafe.Pointer(cUser)) + defer C.free(unsafe.Pointer(cFilePath)) + + cOk, err := C.permission_to_read(cUser, cFilePath) + if cOk == 1 { + return true, nil + } + + if err != nil { + // err contains errno message + return false, err + } + + return false, errors.New("User without permission to read file") +} diff --git a/internal/io/fs/permissions/permission_linuxacl.h b/internal/io/fs/permissions/permission_linuxacl.h new file mode 100644 index 0000000..52dadcf --- /dev/null +++ b/internal/io/fs/permissions/permission_linuxacl.h @@ -0,0 +1,62 @@ +// +build linuxacl + +#ifndef PERMISSION_LINUX_H +#define PERMISSION_LINUX_H + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +//#define DEBUG +#define USER_CHECK 0 +#define GROUP_CHECK 1 + +struct permission_checker { + char *user_name; + uid_t uid; + gid_t *gids; + int ngids; + char *file_path; + struct stat file_stat; + struct passwd pw; +}; + + +#ifdef DEBUG +// Print out permission_checker struct. +void debug_print_checker(struct permission_checker *pc); +#endif + +// Stat a given file to retrieve traditional UNIX permissions. +int stat_file(struct permission_checker *pc); + +// Retrieve UID of user. +int get_user_uid(struct permission_checker *pc); + +// Retrieve all groups of the user. +int get_user_groups(struct permission_checker *pc); + +// Check whether user is member of a group or not. +int is_member_of_group(struct permission_checker *pc, gid_t gid); + +// Check whether user can read file according Linux ACLs. +// As flag use either USER_CHECK or GROUP_CHECK. +int check_acl(struct permission_checker *pc, const int flag); + +// Check whether user has permissions to read file according traditional +// UNIX permissions. As flag use either USER_CHECK or GROUP_CHECK. +int check_traditional(struct permission_checker *pc, const int flag); + +// Returns 1 if user has permission to read file. +// Returns <0 on error and returns 0 if no permissions. +int permission_to_read(char* user, char *file_path); + +#endif // PERMISSION_LINUX_H diff --git a/internal/io/fs/permissions/permission_test.go b/internal/io/fs/permissions/permission_test.go index d415ac2..c0ef038 100644 --- a/internal/io/fs/permissions/permission_test.go +++ b/internal/io/fs/permissions/permission_test.go @@ -1,4 +1,4 @@ -// +build linux +// +build linuxacl package permissions -- cgit v1.2.3 From f46abf36e80a67d13d12dbe4ba8c899026e53961 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 31 Jul 2021 18:20:03 +0300 Subject: more on configurable colors --- internal/io/logger/logger.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) (limited to 'internal/io') diff --git a/internal/io/logger/logger.go b/internal/io/logger/logger.go index 4254eef..bef5293 100644 --- a/internal/io/logger/logger.go +++ b/internal/io/logger/logger.go @@ -12,7 +12,7 @@ import ( "syscall" "time" - "github.com/mimecast/dtail/internal/color" + "github.com/mimecast/dtail/internal/color/brush" "github.com/mimecast/dtail/internal/config" ) @@ -207,8 +207,8 @@ func write(what, severity, message string) { if Mode.logToStdout { line := fmt.Sprintf("%s|%s|%s|%s\n", what, hostname, severity, message) - if color.Colored { - line = color.Colorfy(line) + if config.Client.TermColorsEnabled { + line = brush.Colorfy(line) } stdoutBufCh <- line @@ -262,8 +262,8 @@ func Raw(message string) { } if Mode.logToStdout { - if color.Colored { - message = color.Colorfy(message) + if config.Client.TermColorsEnabled { + message = brush.Colorfy(message) } stdoutBufCh <- message } -- cgit v1.2.3 From 3cc8887885f24a3f0d607af24197bc364ab16b8d Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Thu, 12 Aug 2021 10:56:36 +0300 Subject: add missing brush and also add color client configs plus jsonschema --- internal/io/logger/logger.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'internal/io') diff --git a/internal/io/logger/logger.go b/internal/io/logger/logger.go index bef5293..bb9dc02 100644 --- a/internal/io/logger/logger.go +++ b/internal/io/logger/logger.go @@ -207,7 +207,7 @@ func write(what, severity, message string) { if Mode.logToStdout { line := fmt.Sprintf("%s|%s|%s|%s\n", what, hostname, severity, message) - if config.Client.TermColorsEnabled { + if config.Client.TermColorsEnable { line = brush.Colorfy(line) } @@ -262,7 +262,7 @@ func Raw(message string) { } if Mode.logToStdout { - if config.Client.TermColorsEnabled { + if config.Client.TermColorsEnable { message = brush.Colorfy(message) } stdoutBufCh <- message -- cgit v1.2.3 From c2522ffb59514443816a96386c16bb7527cbe57c Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 21 Aug 2021 14:54:24 +0300 Subject: read files bytewise for more control of whats happening - change transport protocol for more control over newlines --- internal/io/fs/readfile.go | 71 +++++++++++++++++--------------------------- internal/io/logger/logger.go | 7 +++-- 2 files changed, 33 insertions(+), 45 deletions(-) (limited to 'internal/io') diff --git a/internal/io/fs/readfile.go b/internal/io/fs/readfile.go index 6757bd6..8a365a1 100644 --- a/internal/io/fs/readfile.go +++ b/internal/io/fs/readfile.go @@ -14,6 +14,7 @@ import ( "github.com/mimecast/dtail/internal/io/line" "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/protocol" "github.com/mimecast/dtail/internal/regex" "github.com/DataDog/zstd" @@ -148,80 +149,64 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan []byte, t if err != nil { return err } - rawLine := make([]byte, 0, 512) lineLengthThreshold := 1024 * 1024 // 1mb - longLineWarning := false + warnedAboutLongLine := false + message := make([]byte, 0, 512) for { select { case <-ctx.Done(): return nil - default: - } - - select { case <-truncate: if isTruncated, err := f.truncated(fd); isTruncated { return err } - logger.Info(f.filePath, "Current offset", offset) default: } - // Read some bytes (max 4k at once as of go 1.12). isPrefix will - // be set if line does not fit into 4k buffer. - bytes, isPrefix, err := reader.ReadLine() + b, err := reader.ReadByte() if err != nil { - // If EOF, sleep a couple of ms and return with nil error. - // If other error, return with non-nil error. if err != io.EOF { return err } if !f.seekEOF { - logger.Debug(f.FilePath(), "End of file reached") + logger.Info(f.FilePath(), "End of file reached") return nil } time.Sleep(time.Millisecond * 100) continue } + offset++ - rawLine = append(rawLine, bytes...) - offset += uint64(len(bytes)) - - if !isPrefix { - // last LineRead call returned contend until end of line. - rawLine = append(rawLine, '\n') - select { - case rawLines <- rawLine: - case <-ctx.Done(): - return nil + switch b { + case '\n': + if len(message) == 0 { + time.Sleep(time.Millisecond * 100) + continue } - rawLine = make([]byte, 0, 512) - if longLineWarning { - longLineWarning = false - } - continue - } - - // Last LineRead call could not read content until end of line, buffer - // was too small. Determine whether we exceed the max line length we - // want dtail to send to the client at once. Possibly split up log line - // into multiple log lines. - if len(rawLine) >= lineLengthThreshold { - if !longLineWarning { - f.serverMessages <- logger.Warn(f.filePath, "Long log line, splitting into multiple lines") - // Only print out one warning per long log line. - longLineWarning = true - } - rawLine = append(rawLine, '\n') select { - case rawLines <- rawLine: + case rawLines <- append(message, protocol.MessageDelimiter): + message = make([]byte, 0, 512) + warnedAboutLongLine = false case <-ctx.Done(): return nil } - rawLine = make([]byte, 0, 512) + default: + if len(message) >= lineLengthThreshold { + if !warnedAboutLongLine { + f.serverMessages <- logger.Warn(f.filePath, "Long log line, splitting into multiple lines") + warnedAboutLongLine = true + } + select { + case <-ctx.Done(): + return nil + case rawLines <- append(message, protocol.MessageDelimiter): + message = make([]byte, 0, 512) + } + } + message = append(message, b) } } } diff --git a/internal/io/logger/logger.go b/internal/io/logger/logger.go index bb9dc02..3a3935d 100644 --- a/internal/io/logger/logger.go +++ b/internal/io/logger/logger.go @@ -205,7 +205,7 @@ func Trace(args ...interface{}) string { // Write log line to buffer and/or log file. func write(what, severity, message string) { if Mode.logToStdout { - line := fmt.Sprintf("%s|%s|%s|%s\n", what, hostname, severity, message) + line := fmt.Sprintf("%s|%s|%s|%s", what, hostname, severity, message) if config.Client.TermColorsEnable { line = brush.Colorfy(line) @@ -219,7 +219,7 @@ func write(what, severity, message string) { timeStr := t.Format("20060102-150405") fileLogBufCh <- buf{ time: t, - message: fmt.Sprintf("%s|%s|%s|%s\n", severity, timeStr, what, message), + message: fmt.Sprintf("%s|%s|%s|%s", severity, timeStr, what, message), } } } @@ -326,6 +326,7 @@ func Flush() { select { case message := <-stdoutBufCh: stdoutWriter.WriteString(message) + stdoutWriter.WriteString("\n") default: stdoutWriter.Flush() return @@ -338,6 +339,7 @@ func writeToStdout(ctx context.Context) { select { case message := <-stdoutBufCh: stdoutWriter.WriteString(message) + stdoutWriter.WriteString("\n") case <-time.After(time.Millisecond * 100): stdoutWriter.Flush() case <-pauseCh: @@ -365,6 +367,7 @@ func writeToFile(ctx context.Context) { dateStr := buf.time.Format("20060102") w := fileWriter(dateStr) w.WriteString(buf.message) + w.WriteString("\n") case <-pauseCh: PAUSE: for { -- cgit v1.2.3 From 6d727b9bdbc387c8a5c34406a2c4de9140face38 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 28 Aug 2021 19:36:46 +0100 Subject: use a byte.Buffer in the file reader --- internal/io/fs/readfile.go | 32 +++++++++++++++++++------------- internal/io/line/line.go | 5 +++-- internal/io/pool/bytesbuffer.go | 19 +++++++++++++++++++ 3 files changed, 41 insertions(+), 15 deletions(-) create mode 100644 internal/io/pool/bytesbuffer.go (limited to 'internal/io') diff --git a/internal/io/fs/readfile.go b/internal/io/fs/readfile.go index 8a365a1..e44f30e 100644 --- a/internal/io/fs/readfile.go +++ b/internal/io/fs/readfile.go @@ -2,6 +2,7 @@ package fs import ( "bufio" + "bytes" "compress/gzip" "context" "errors" @@ -14,6 +15,7 @@ import ( "github.com/mimecast/dtail/internal/io/line" "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/pool" "github.com/mimecast/dtail/internal/protocol" "github.com/mimecast/dtail/internal/regex" @@ -90,7 +92,7 @@ func (f readFile) Start(ctx context.Context, lines chan<- line.Line, re regex.Re fd.Seek(0, io.SeekEnd) } - rawLines := make(chan []byte, 100) + rawLines := make(chan *bytes.Buffer, 100) truncate := make(chan struct{}) var wg sync.WaitGroup @@ -142,7 +144,7 @@ func (f readFile) makeReader(fd *os.File) (reader *bufio.Reader, err error) { return } -func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan []byte, truncate <-chan struct{}) error { +func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan *bytes.Buffer, truncate <-chan struct{}) error { var offset uint64 reader, err := f.makeReader(fd) @@ -152,7 +154,7 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan []byte, t lineLengthThreshold := 1024 * 1024 // 1mb warnedAboutLongLine := false - message := make([]byte, 0, 512) + message := pool.BytesBuffer.Get().(*bytes.Buffer) for { select { @@ -182,37 +184,41 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan []byte, t switch b { case '\n': - if len(message) == 0 { + if message.Len() == 0 { time.Sleep(time.Millisecond * 100) continue } + message.WriteByte(protocol.MessageDelimiter) select { - case rawLines <- append(message, protocol.MessageDelimiter): - message = make([]byte, 0, 512) + case rawLines <- message: + message = pool.BytesBuffer.Get().(*bytes.Buffer) + //fmt.Printf("%d %d %p\n", message.Len(), message.Cap(), message) warnedAboutLongLine = false case <-ctx.Done(): return nil } default: - if len(message) >= lineLengthThreshold { + if message.Len() >= lineLengthThreshold { if !warnedAboutLongLine { f.serverMessages <- logger.Warn(f.filePath, "Long log line, splitting into multiple lines") warnedAboutLongLine = true } + message.WriteByte(protocol.MessageDelimiter) select { + case rawLines <- message: + message = pool.BytesBuffer.Get().(*bytes.Buffer) + //fmt.Printf("%d %d %p\n", message.Len(), message.Cap(), message) case <-ctx.Done(): return nil - case rawLines <- append(message, protocol.MessageDelimiter): - message = make([]byte, 0, 512) } } - message = append(message, b) + message.WriteByte(b) } } } // Filter log lines matching a given regular expression. -func (f readFile) filter(ctx context.Context, wg *sync.WaitGroup, rawLines <-chan []byte, lines chan<- line.Line, re regex.Regex) { +func (f readFile) filter(ctx context.Context, wg *sync.WaitGroup, rawLines <-chan *bytes.Buffer, lines chan<- line.Line, re regex.Regex) { defer wg.Done() for { @@ -233,10 +239,10 @@ func (f readFile) filter(ctx context.Context, wg *sync.WaitGroup, rawLines <-cha } } -func (f readFile) transmittable(lineBytes []byte, length, capacity int, re regex.Regex) (line.Line, bool) { +func (f readFile) transmittable(lineBytes *bytes.Buffer, length, capacity int, re regex.Regex) (line.Line, bool) { var read line.Line - if !re.Match(lineBytes) { + if !re.Match(lineBytes.Bytes()) { f.updateLineNotMatched() f.updateLineNotTransmitted() return read, false diff --git a/internal/io/line/line.go b/internal/io/line/line.go index 715be34..d306c88 100644 --- a/internal/io/line/line.go +++ b/internal/io/line/line.go @@ -1,13 +1,14 @@ package line import ( + "bytes" "fmt" ) // Line represents a read log line. type Line struct { // The content of the log line. - Content []byte + Content *bytes.Buffer // Until now, how many log lines were processed? Count uint64 // Sometimes we produce too many log lines so that the client @@ -25,7 +26,7 @@ type Line struct { // Return a human readable representation of the followed line. func (l Line) String() string { return fmt.Sprintf("Line(Content:%s,TransmittedPerc:%v,Count:%v,SourceID:%s)", - string(l.Content), + l.Content.String(), l.TransmittedPerc, l.Count, l.SourceID) diff --git a/internal/io/pool/bytesbuffer.go b/internal/io/pool/bytesbuffer.go new file mode 100644 index 0000000..0a159f5 --- /dev/null +++ b/internal/io/pool/bytesbuffer.go @@ -0,0 +1,19 @@ +package pool + +import ( + "bytes" + "sync" +) + +var BytesBuffer = sync.Pool{ + New: func() interface{} { + b := bytes.Buffer{} + b.Grow(128) + return &b + }, +} + +func RecycleBytesBuffer(b *bytes.Buffer) { + b.Reset() + BytesBuffer.Put(b) +} -- cgit v1.2.3 From 23982f331c2154a66b86d596226c24454fd06be5 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 28 Aug 2021 20:26:32 +0100 Subject: 1. Major performance gain by not checking for file truncation aftter each bytes read. 2. Introduce field separator to the protocol package. --- internal/io/fs/readfile.go | 20 +++++++++----------- 1 file changed, 9 insertions(+), 11 deletions(-) (limited to 'internal/io') diff --git a/internal/io/fs/readfile.go b/internal/io/fs/readfile.go index e44f30e..f2f672a 100644 --- a/internal/io/fs/readfile.go +++ b/internal/io/fs/readfile.go @@ -157,22 +157,21 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan *bytes.Bu message := pool.BytesBuffer.Get().(*bytes.Buffer) for { - select { - case <-ctx.Done(): - return nil - case <-truncate: - if isTruncated, err := f.truncated(fd); isTruncated { - return err - } - default: - } - b, err := reader.ReadByte() if err != nil { if err != io.EOF { return err } + select { + case <-truncate: + if isTruncated, err := f.truncated(fd); isTruncated { + return err + } + case <-ctx.Done(): + return nil + default: + } if !f.seekEOF { logger.Info(f.FilePath(), "End of file reached") return nil @@ -207,7 +206,6 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan *bytes.Bu select { case rawLines <- message: message = pool.BytesBuffer.Get().(*bytes.Buffer) - //fmt.Printf("%d %d %p\n", message.Len(), message.Cap(), message) case <-ctx.Done(): return nil } -- cgit v1.2.3 From f74a9e4b35feb8c07d8a70b5a581088a0a59889d Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Tue, 7 Sep 2021 10:01:32 +0300 Subject: Produce MAPREDUCE lines, can aggregate these via default log format --- internal/io/logger/logger.go | 40 ++++++++++++++++++++++++++++++++-------- internal/io/pool/builder.go | 18 ++++++++++++++++++ 2 files changed, 50 insertions(+), 8 deletions(-) create mode 100644 internal/io/pool/builder.go (limited to 'internal/io') diff --git a/internal/io/logger/logger.go b/internal/io/logger/logger.go index 3a3935d..6890201 100644 --- a/internal/io/logger/logger.go +++ b/internal/io/logger/logger.go @@ -14,6 +14,8 @@ import ( "github.com/mimecast/dtail/internal/color/brush" "github.com/mimecast/dtail/internal/config" + "github.com/mimecast/dtail/internal/io/pool" + "github.com/mimecast/dtail/internal/protocol" ) const ( @@ -132,6 +134,24 @@ func Info(args ...interface{}) string { return log(clientStr, infoStr, args) } +// Mapreduce message logging. +func Mapreduce(table string, data map[string]interface{}) string { + args := make([]interface{}, len(data)+1) + + args[0] = fmt.Sprintf("MAPREDUCE:%s", strings.ToUpper(table)) + i := 1 + for k, v := range data { + args[i] = fmt.Sprintf("%s=%v", k, v) + i++ + } + + if Mode.Server { + return log(serverStr, infoStr, args) + } + + return log(clientStr, infoStr, args) +} + // Warn message logging. func Warn(args ...interface{}) string { if !Mode.Quiet { @@ -230,24 +250,28 @@ func log(what string, severity string, args []interface{}) string { return "" } - messages := []string{} + sb := pool.BuilderBuffer.Get().(*strings.Builder) + + for i, arg := range args { + if i > 0 { + sb.WriteString(protocol.FieldDelimiter) + } - for _, arg := range args { switch v := arg.(type) { case string: - messages = append(messages, v) + sb.WriteString(v) case int: - messages = append(messages, fmt.Sprintf("%d", v)) + sb.WriteString(fmt.Sprintf("%d", v)) case error: - messages = append(messages, v.Error()) + sb.WriteString(v.Error()) default: - messages = append(messages, fmt.Sprintf("%v", v)) + sb.WriteString(fmt.Sprintf("%v", v)) } } - message := strings.Join(messages, "|") + message := sb.String() + pool.RecycleBuilderBuffer(sb) write(what, severity, message) - return fmt.Sprintf("%s|%s", severity, message) } diff --git a/internal/io/pool/builder.go b/internal/io/pool/builder.go new file mode 100644 index 0000000..c9dc221 --- /dev/null +++ b/internal/io/pool/builder.go @@ -0,0 +1,18 @@ +package pool + +import ( + "strings" + "sync" +) + +var BuilderBuffer = sync.Pool{ + New: func() interface{} { + sb := strings.Builder{} + return &sb + }, +} + +func RecycleBuilderBuffer(sb *strings.Builder) { + sb.Reset() + BuilderBuffer.Put(sb) +} -- cgit v1.2.3 From 16dc57e1e1c28e9d762424e596223a980770e059 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Wed, 8 Sep 2021 19:10:50 +0300 Subject: mapreduce tables are in colors now too --- internal/io/fs/readfile.go | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) (limited to 'internal/io') diff --git a/internal/io/fs/readfile.go b/internal/io/fs/readfile.go index f2f672a..c0d44dd 100644 --- a/internal/io/fs/readfile.go +++ b/internal/io/fs/readfile.go @@ -16,7 +16,6 @@ import ( "github.com/mimecast/dtail/internal/io/line" "github.com/mimecast/dtail/internal/io/logger" "github.com/mimecast/dtail/internal/io/pool" - "github.com/mimecast/dtail/internal/protocol" "github.com/mimecast/dtail/internal/regex" "github.com/DataDog/zstd" @@ -187,7 +186,7 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan *bytes.Bu time.Sleep(time.Millisecond * 100) continue } - message.WriteByte(protocol.MessageDelimiter) + message.WriteString("\n") select { case rawLines <- message: message = pool.BytesBuffer.Get().(*bytes.Buffer) @@ -202,7 +201,7 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan *bytes.Bu f.serverMessages <- logger.Warn(f.filePath, "Long log line, splitting into multiple lines") warnedAboutLongLine = true } - message.WriteByte(protocol.MessageDelimiter) + message.WriteString("\n") select { case rawLines <- message: message = pool.BytesBuffer.Get().(*bytes.Buffer) -- cgit v1.2.3 From 6506e20f6c80f4acb7434eb9dd14f784a67189cd Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 18 Sep 2021 14:41:25 +0300 Subject: add spartan mode --- internal/io/fs/readfile.go | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) (limited to 'internal/io') diff --git a/internal/io/fs/readfile.go b/internal/io/fs/readfile.go index c0d44dd..ec33c60 100644 --- a/internal/io/fs/readfile.go +++ b/internal/io/fs/readfile.go @@ -182,11 +182,14 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan *bytes.Bu switch b { case '\n': - if message.Len() == 0 { - time.Sleep(time.Millisecond * 100) - continue - } - message.WriteString("\n") + /* + // dcat/dgrep should actually transfer empty lines + if message.Len() == 0 { + time.Sleep(time.Millisecond * 100) + continue + } + */ + //message.WriteString("\n") select { case rawLines <- message: message = pool.BytesBuffer.Get().(*bytes.Buffer) -- cgit v1.2.3 From fe3e68afd99d8ea246be52893730f987e138ec24 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sun, 19 Sep 2021 13:22:59 +0300 Subject: move args to config package logger package rewrite as dlog --- internal/io/dlog/dlog.go | 206 +++++++++++++++++++++++++++++++ internal/io/dlog/level.go | 89 +++++++++++++ internal/io/dlog/loggers/factory.go | 60 +++++++++ internal/io/dlog/loggers/file.go | 156 +++++++++++++++++++++++ internal/io/dlog/loggers/fout.go | 46 +++++++ internal/io/dlog/loggers/logger.go | 18 +++ internal/io/dlog/loggers/none.go | 21 ++++ internal/io/dlog/loggers/stdout.go | 73 +++++++++++ internal/io/dlog/rotation.go | 27 ++++ internal/io/dlog/source.go | 19 +++ internal/io/dlog/strategy.go | 22 ++++ internal/io/fs/permissions/permission.go | 4 +- internal/io/fs/readfile.go | 16 +-- internal/io/logger/logger.go | 23 +++- internal/io/logger/modes.go | 9 +- internal/io/prompt/prompt.go | 6 +- 16 files changed, 774 insertions(+), 21 deletions(-) create mode 100644 internal/io/dlog/dlog.go create mode 100644 internal/io/dlog/level.go create mode 100644 internal/io/dlog/loggers/factory.go create mode 100644 internal/io/dlog/loggers/file.go create mode 100644 internal/io/dlog/loggers/fout.go create mode 100644 internal/io/dlog/loggers/logger.go create mode 100644 internal/io/dlog/loggers/none.go create mode 100644 internal/io/dlog/loggers/stdout.go create mode 100644 internal/io/dlog/rotation.go create mode 100644 internal/io/dlog/source.go create mode 100644 internal/io/dlog/strategy.go (limited to 'internal/io') diff --git a/internal/io/dlog/dlog.go b/internal/io/dlog/dlog.go new file mode 100644 index 0000000..7282741 --- /dev/null +++ b/internal/io/dlog/dlog.go @@ -0,0 +1,206 @@ +package dlog + +import ( + "context" + "fmt" + "strings" + "sync" + "time" + + "github.com/mimecast/dtail/internal/color/brush" + "github.com/mimecast/dtail/internal/config" + "github.com/mimecast/dtail/internal/io/dlog/loggers" + "github.com/mimecast/dtail/internal/io/pool" + "github.com/mimecast/dtail/internal/protocol" +) + +// Client is the log handler for the client packages. +var Client *DLog + +// Server is the log handler for the server packages. +var Server *DLog + +// Common is the log handler for all other packages. +// TODO: Rename Common to Common +var Common *DLog + +var mutex sync.Mutex +var started bool + +// Start logger(s). +func Start(ctx context.Context, wg *sync.WaitGroup, sourceProcess source, logLevel string) { + mutex.Lock() + defer mutex.Unlock() + + if started { + Common.FatalPanic("Logger already started") + } + + level := newLevel(logLevel) + switch sourceProcess { + case CLIENT: + // This is a DTail client process running. + impl := loggers.FOUT + Client = New(CLIENT, CLIENT, impl, level) + Server = New(CLIENT, SER