refactor process log follow into helpers
- commit
- aa1feb182f5f7e9ff12008df7ca2fcd47d8cc3af
- author
- brice · · 2026-02-08 17:17 UTC
- parent
- c0ac371
- signature
- unsigned
1 file · +115 -66
M internal/usecase/logs/service.go +115 -66
| @@ -86,85 +86,134 @@ func (s *Service) FollowProcessLogs(ctx context.Context, initialLines int) (<-ch | |||
| 86 | 86 | zerowrap.FieldUseCase: "FollowProcessLogs", | |
| 87 | 87 | "initial_lines": initialLines, | |
| 88 | 88 | }) | |
| 89 | + | if !s.useFileProcessLogs() { | |
| 90 | + | return s.followProcessLogsFromJournal(ctx, initialLines) | |
| 91 | + | } | |
| 92 | + | ||
| 93 | + | return s.followProcessLogsFromFileOrJournal(ctx, initialLines) | |
| 94 | + | } | |
| 95 | + | ||
| 96 | + | func (s *Service) useFileProcessLogs() bool { | |
| 97 | + | return s.fileLoggingEnabled | |
| 98 | + | } | |
| 99 | + | ||
| 100 | + | func (s *Service) followProcessLogsFromFileOrJournal( | |
| 101 | + | ctx context.Context, | |
| 102 | + | initialLines int, | |
| 103 | + | ) (<-chan string, error) { | |
| 89 | 104 | log := zerowrap.FromCtx(ctx) | |
| 90 | 105 | ||
| 91 | - | if s.useFileProcessLogs() { | |
| 92 | - | if s.logFilePath == "" { | |
| 93 | - | return nil, fmt.Errorf("log file path not configured") | |
| 106 | + | file, err := s.openProcessLogFile() | |
| 107 | + | if err != nil { | |
| 108 | + | if os.IsNotExist(err) { | |
| 109 | + | log.Warn().Str("log_path", s.logFilePath). | |
| 110 | + | Msg("process log file missing, falling back to journalctl follow") | |
| 111 | + | return s.followProcessLogsFromJournal(ctx, initialLines) | |
| 94 | 112 | } | |
| 113 | + | return nil, log.WrapErr(err, "failed to open log file") | |
| 114 | + | } | |
| 95 | 115 | ||
| 96 | - | file, err := os.Open(s.logFilePath) | |
| 97 | - | if err != nil { | |
| 98 | - | if os.IsNotExist(err) { | |
| 99 | - | log.Warn().Str("log_path", s.logFilePath). | |
| 100 | - | Msg("process log file missing, falling back to journalctl follow") | |
| 101 | - | return s.followProcessLogsFromJournal(ctx, initialLines) | |
| 102 | - | } | |
| 103 | - | return nil, log.WrapErr(err, "failed to open log file") | |
| 116 | + | return s.followProcessLogsFromFile(ctx, file, initialLines), nil | |
| 117 | + | } | |
| 118 | + | ||
| 119 | + | func (s *Service) openProcessLogFile() (*os.File, error) { | |
| 120 | + | if s.logFilePath == "" { | |
| 121 | + | return nil, fmt.Errorf("log file path not configured") | |
| 122 | + | } | |
| 123 | + | ||
| 124 | + | return os.Open(s.logFilePath) | |
| 125 | + | } | |
| 126 | + | ||
| 127 | + | func (s *Service) followProcessLogsFromFile( | |
| 128 | + | ctx context.Context, | |
| 129 | + | file *os.File, | |
| 130 | + | initialLines int, | |
| 131 | + | ) <-chan string { | |
| 132 | + | log := zerowrap.FromCtx(ctx) | |
| 133 | + | ch := make(chan string, 100) | |
| 134 | + | ||
| 135 | + | go func() { | |
| 136 | + | defer close(ch) | |
| 137 | + | defer file.Close() | |
| 138 | + | ||
| 139 | + | if !s.emitInitialFileLogs(ctx, file, initialLines, ch) { | |
| 140 | + | return | |
| 141 | + | } | |
| 142 | + | if err := s.seekToEndForFileFollow(file); err != nil { | |
| 143 | + | log.Warn().Err(err).Msg("failed to seek to end") | |
| 144 | + | return | |
| 104 | 145 | } | |
| 105 | 146 | ||
| 106 | - | ch := make(chan string, 100) | |
| 107 | - | ||
| 108 | - | go func() { | |
| 109 | - | defer close(ch) | |
| 110 | - | defer file.Close() | |
| 111 | - | ||
| 112 | - | // First, get initial lines | |
| 113 | - | if initialLines > 0 { | |
| 114 | - | lines, err := tailLines(file, initialLines) | |
| 115 | - | if err != nil { | |
| 116 | - | log.Warn().Err(err).Msg("failed to read initial lines") | |
| 117 | - | } else { | |
| 118 | - | for _, line := range lines { | |
| 119 | - | select { | |
| 120 | - | case ch <- line: | |
| 121 | - | case <-ctx.Done(): | |
| 122 | - | return | |
| 123 | - | } | |
| 124 | - | } | |
| 125 | - | } | |
| 126 | - | } | |
| 147 | + | s.streamFileFollowLogs(ctx, file, ch) | |
| 148 | + | }() | |
| 127 | 149 | ||
| 128 | - | // Seek to end for following | |
| 129 | - | if _, err := file.Seek(0, io.SeekEnd); err != nil { | |
| 130 | - | log.Warn().Err(err).Msg("failed to seek to end") | |
| 131 | - | return | |
| 132 | - | } | |
| 150 | + | return ch | |
| 151 | + | } | |
| 133 | 152 | ||
| 134 | - | // Follow new lines | |
| 135 | - | reader := bufio.NewReader(file) | |
| 136 | - | for { | |
| 137 | - | select { | |
| 138 | - | case <-ctx.Done(): | |
| 139 | - | return | |
| 140 | - | default: | |
| 141 | - | line, err := reader.ReadString('\n') | |
| 142 | - | if err != nil { | |
| 143 | - | if err == io.EOF { | |
| 144 | - | time.Sleep(100 * time.Millisecond) | |
| 145 | - | continue | |
| 146 | - | } | |
| 147 | - | log.Warn().Err(err).Msg("error reading log file") | |
| 148 | - | return | |
| 149 | - | } | |
| 150 | - | line = strings.TrimRight(line, "\n\r") | |
| 151 | - | select { | |
| 152 | - | case ch <- line: | |
| 153 | - | case <-ctx.Done(): | |
| 154 | - | return | |
| 155 | - | } | |
| 156 | - | } | |
| 157 | - | } | |
| 158 | - | }() | |
| 153 | + | func (s *Service) emitInitialFileLogs( | |
| 154 | + | ctx context.Context, | |
| 155 | + | file *os.File, | |
| 156 | + | initialLines int, | |
| 157 | + | ch chan<- string, | |
| 158 | + | ) bool { | |
| 159 | + | if initialLines <= 0 { | |
| 160 | + | return true | |
| 161 | + | } | |
| 162 | + | ||
| 163 | + | log := zerowrap.FromCtx(ctx) | |
| 164 | + | lines, err := tailLines(file, initialLines) | |
| 165 | + | if err != nil { | |
| 166 | + | log.Warn().Err(err).Msg("failed to read initial lines") | |
| 167 | + | return true | |
| 168 | + | } | |
| 159 | 169 | ||
| 160 | - | return ch, nil | |
| 170 | + | for _, line := range lines { | |
| 171 | + | if !sendLogLineWithChannel(ctx, ch, line) { | |
| 172 | + | return false | |
| 173 | + | } | |
| 161 | 174 | } | |
| 162 | 175 | ||
| 163 | - | return s.followProcessLogsFromJournal(ctx, initialLines) | |
| 176 | + | return true | |
| 164 | 177 | } | |
| 165 | 178 | ||
| 166 | - | func (s *Service) useFileProcessLogs() bool { | |
| 167 | - | return s.fileLoggingEnabled | |
| 179 | + | func (s *Service) seekToEndForFileFollow(file *os.File) error { | |
| 180 | + | _, err := file.Seek(0, io.SeekEnd) | |
| 181 | + | return err | |
| 182 | + | } | |
| 183 | + | ||
| 184 | + | func (s *Service) streamFileFollowLogs(ctx context.Context, file *os.File, ch chan<- string) { | |
| 185 | + | log := zerowrap.FromCtx(ctx) | |
| 186 | + | reader := bufio.NewReader(file) | |
| 187 | + | ||
| 188 | + | for { | |
| 189 | + | if ctx.Err() != nil { | |
| 190 | + | return | |
| 191 | + | } | |
| 192 | + | ||
| 193 | + | line, err := reader.ReadString('\n') | |
| 194 | + | if err != nil { | |
| 195 | + | if err == io.EOF { | |
| 196 | + | time.Sleep(100 * time.Millisecond) | |
| 197 | + | continue | |
| 198 | + | } | |
| 199 | + | log.Warn().Err(err).Msg("error reading log file") | |
| 200 | + | return | |
| 201 | + | } | |
| 202 | + | ||
| 203 | + | line = strings.TrimRight(line, "\n\r") | |
| 204 | + | if !sendLogLineWithChannel(ctx, ch, line) { | |
| 205 | + | return | |
| 206 | + | } | |
| 207 | + | } | |
| 208 | + | } | |
| 209 | + | ||
| 210 | + | func sendLogLineWithChannel(ctx context.Context, ch chan<- string, line string) bool { | |
| 211 | + | select { | |
| 212 | + | case ch <- line: | |
| 213 | + | return true | |
| 214 | + | case <-ctx.Done(): | |
| 215 | + | return false | |
| 216 | + | } | |
| 168 | 217 | } | |
| 169 | 218 | ||
| 170 | 219 | func (s *Service) getProcessLogsFromJournal(ctx context.Context, lines int) ([]string, error) { | |
| @@ -86,85 +86,134 @@ func (s *Service) FollowProcessLogs(ctx context.Context, initialLines int) (<-ch | |||||
| 86 | zerowrap.FieldUseCase: "FollowProcessLogs", | 86 | zerowrap.FieldUseCase: "FollowProcessLogs", | ||
| 87 | "initial_lines": initialLines, | 87 | "initial_lines": initialLines, | ||
| 88 | }) | 88 | }) | ||
| 89 | + | if !s.useFileProcessLogs() { | |||
| 90 | + | return s.followProcessLogsFromJournal(ctx, initialLines) | |||
| 91 | + | } | |||
| 92 | + | ||||
| 93 | + | return s.followProcessLogsFromFileOrJournal(ctx, initialLines) | |||
| 94 | + | } | |||
| 95 | + | ||||
| 96 | + | func (s *Service) useFileProcessLogs() bool { | |||
| 97 | + | return s.fileLoggingEnabled | |||
| 98 | + | } | |||
| 99 | + | ||||
| 100 | + | func (s *Service) followProcessLogsFromFileOrJournal( | |||
| 101 | + | ctx context.Context, | |||
| 102 | + | initialLines int, | |||
| 103 | + | ) (<-chan string, error) { | |||
| 89 | log := zerowrap.FromCtx(ctx) | 104 | log := zerowrap.FromCtx(ctx) | ||
| 90 | 105 | ||||
| 91 | - | if s.useFileProcessLogs() { | 106 | + | file, err := s.openProcessLogFile() |
| 92 | - | if s.logFilePath == "" { | 107 | + | if err != nil { |
| 93 | - | return nil, fmt.Errorf("log file path not configured") | 108 | + | if os.IsNotExist(err) { |
| 109 | + | log.Warn().Str("log_path", s.logFilePath). | |||
| 110 | + | Msg("process log file missing, falling back to journalctl follow") | |||
| 111 | + | return s.followProcessLogsFromJournal(ctx, initialLines) | |||
| 94 | } | 112 | } | ||
| 113 | + | return nil, log.WrapErr(err, "failed to open log file") | |||
| 114 | + | } | |||
| 95 | 115 | ||||
| 96 | - | file, err := os.Open(s.logFilePath) | 116 | + | return s.followProcessLogsFromFile(ctx, file, initialLines), nil |
| 97 | - | if err != nil { | 117 | + | } |
| 98 | - | if os.IsNotExist(err) { | 118 | + | |
| 99 | - | log.Warn().Str("log_path", s.logFilePath). | 119 | + | func (s *Service) openProcessLogFile() (*os.File, error) { |
| 100 | - | Msg("process log file missing, falling back to journalctl follow") | 120 | + | if s.logFilePath == "" { |
| 101 | - | return s.followProcessLogsFromJournal(ctx, initialLines) | 121 | + | return nil, fmt.Errorf("log file path not configured") |
| 102 | - | } | 122 | + | } |
| 103 | - | return nil, log.WrapErr(err, "failed to open log file") | 123 | + | |
| 124 | + | return os.Open(s.logFilePath) | |||
| 125 | + | } | |||
| 126 | + | ||||
| 127 | + | func (s *Service) followProcessLogsFromFile( | |||
| 128 | + | ctx context.Context, | |||
| 129 | + | file *os.File, | |||
| 130 | + | initialLines int, | |||
| 131 | + | ) <-chan string { | |||
| 132 | + | log := zerowrap.FromCtx(ctx) | |||
| 133 | + | ch := make(chan string, 100) | |||
| 134 | + | ||||
| 135 | + | go func() { | |||
| 136 | + | defer close(ch) | |||
| 137 | + | defer file.Close() | |||
| 138 | + | ||||
| 139 | + | if !s.emitInitialFileLogs(ctx, file, initialLines, ch) { | |||
| 140 | + | return | |||
| 141 | + | } | |||
| 142 | + | if err := s.seekToEndForFileFollow(file); err != nil { | |||
| 143 | + | log.Warn().Err(err).Msg("failed to seek to end") | |||
| 144 | + | return | |||
| 104 | } | 145 | } | ||
| 105 | 146 | ||||
| 106 | - | ch := make(chan string, 100) | 147 | + | s.streamFileFollowLogs(ctx, file, ch) |
| 107 | - | 148 | + | }() | |
| 108 | - | go func() { | |||
| 109 | - | defer close(ch) | |||
| 110 | - | defer file.Close() | |||
| 111 | - | ||||
| 112 | - | // First, get initial lines | |||
| 113 | - | if initialLines > 0 { | |||
| 114 | - | lines, err := tailLines(file, initialLines) | |||
| 115 | - | if err != nil { | |||
| 116 | - | log.Warn().Err(err).Msg("failed to read initial lines") | |||
| 117 | - | } else { | |||
| 118 | - | for _, line := range lines { | |||
| 119 | - | select { | |||
| 120 | - | case ch <- line: | |||
| 121 | - | case <-ctx.Done(): | |||
| 122 | - | return | |||
| 123 | - | } | |||
| 124 | - | } | |||
| 125 | - | } | |||
| 126 | - | } | |||
| 127 | 149 | ||||
| 128 | - | // Seek to end for following | 150 | + | return ch |
| 129 | - | if _, err := file.Seek(0, io.SeekEnd); err != nil { | 151 | + | } |
| 130 | - | log.Warn().Err(err).Msg("failed to seek to end") | |||
| 131 | - | return | |||
| 132 | - | } | |||
| 133 | 152 | ||||
| 134 | - | // Follow new lines | 153 | + | func (s *Service) emitInitialFileLogs( |
| 135 | - | reader := bufio.NewReader(file) | 154 | + | ctx context.Context, |
| 136 | - | for { | 155 | + | file *os.File, |
| 137 | - | select { | 156 | + | initialLines int, |
| 138 | - | case <-ctx.Done(): | 157 | + | ch chan<- string, |
| 139 | - | return | 158 | + | ) bool { |
| 140 | - | default: | 159 | + | if initialLines <= 0 { |
| 141 | - | line, err := reader.ReadString('\n') | 160 | + | return true |
| 142 | - | if err != nil { | 161 | + | } |
| 143 | - | if err == io.EOF { | 162 | + | |
| 144 | - | time.Sleep(100 * time.Millisecond) | 163 | + | log := zerowrap.FromCtx(ctx) |
| 145 | - | continue | 164 | + | lines, err := tailLines(file, initialLines) |
| 146 | - | } | 165 | + | if err != nil { |
| 147 | - | log.Warn().Err(err).Msg("error reading log file") | 166 | + | log.Warn().Err(err).Msg("failed to read initial lines") |
| 148 | - | return | 167 | + | return true |
| 149 | - | } | 168 | + | } |
| 150 | - | line = strings.TrimRight(line, "\n\r") | |||
| 151 | - | select { | |||
| 152 | - | case ch <- line: | |||
| 153 | - | case <-ctx.Done(): | |||
| 154 | - | return | |||
| 155 | - | } | |||
| 156 | - | } | |||
| 157 | - | } | |||
| 158 | - | }() | |||
| 159 | 169 | ||||
| 160 | - | return ch, nil | 170 | + | for _, line := range lines { |
| 171 | + | if !sendLogLineWithChannel(ctx, ch, line) { | |||
| 172 | + | return false | |||
| 173 | + | } | |||
| 161 | } | 174 | } | ||
| 162 | 175 | ||||
| 163 | - | return s.followProcessLogsFromJournal(ctx, initialLines) | 176 | + | return true |
| 164 | } | 177 | } | ||
| 165 | 178 | ||||
| 166 | - | func (s *Service) useFileProcessLogs() bool { | 179 | + | func (s *Service) seekToEndForFileFollow(file *os.File) error { |
| 167 | - | return s.fileLoggingEnabled | 180 | + | _, err := file.Seek(0, io.SeekEnd) |
| 181 | + | return err | |||
| 182 | + | } | |||
| 183 | + | ||||
| 184 | + | func (s *Service) streamFileFollowLogs(ctx context.Context, file *os.File, ch chan<- string) { | |||
| 185 | + | log := zerowrap.FromCtx(ctx) | |||
| 186 | + | reader := bufio.NewReader(file) | |||
| 187 | + | ||||
| 188 | + | for { | |||
| 189 | + | if ctx.Err() != nil { | |||
| 190 | + | return | |||
| 191 | + | } | |||
| 192 | + | ||||
| 193 | + | line, err := reader.ReadString('\n') | |||
| 194 | + | if err != nil { | |||
| 195 | + | if err == io.EOF { | |||
| 196 | + | time.Sleep(100 * time.Millisecond) | |||
| 197 | + | continue | |||
| 198 | + | } | |||
| 199 | + | log.Warn().Err(err).Msg("error reading log file") | |||
| 200 | + | return | |||
| 201 | + | } | |||
| 202 | + | ||||
| 203 | + | line = strings.TrimRight(line, "\n\r") | |||
| 204 | + | if !sendLogLineWithChannel(ctx, ch, line) { | |||
| 205 | + | return | |||
| 206 | + | } | |||
| 207 | + | } | |||
| 208 | + | } | |||
| 209 | + | ||||
| 210 | + | func sendLogLineWithChannel(ctx context.Context, ch chan<- string, line string) bool { | |||
| 211 | + | select { | |||
| 212 | + | case ch <- line: | |||
| 213 | + | return true | |||
| 214 | + | case <-ctx.Done(): | |||
| 215 | + | return false | |||
| 216 | + | } | |||
| 168 | } | 217 | } | ||
| 169 | 218 | ||||
| 170 | func (s *Service) getProcessLogsFromJournal(ctx context.Context, lines int) ([]string, error) { | 219 | func (s *Service) getProcessLogsFromJournal(ctx context.Context, lines int) ([]string, error) { | ||