File: file_logger.go

package info (click to toggle)
golang-github-viant-toolbox 0.33.2-3
  • links: PTS, VCS
  • area: main
  • in suites: bullseye
  • size: 1,280 kB
  • sloc: makefile: 16
file content (297 lines) | stat: -rw-r--r-- 7,339 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
package toolbox

import (
	"bytes"
	"errors"
	"fmt"
	"os"
	"os/signal"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"time"
)

//FileLoggerConfig represents FileLogger
type FileLoggerConfig struct {
	LogType            string
	FileTemplate       string
	filenameProvider   func(t time.Time) string
	QueueFlashCount    int
	MaxQueueSize       int
	FlushRequencyInMs  int //type backward-forward compatibility
	FlushFrequencyInMs int
	MaxIddleTimeInSec  int
	inited             bool
}

func (c *FileLoggerConfig) Init() {
	if c.inited {
		return
	}
	defaultProvider := func(t time.Time) string {
		return c.FileTemplate
	}
	c.inited = true
	template := c.FileTemplate
	c.filenameProvider = defaultProvider
	startIndex := strings.Index(template, "[")
	if startIndex == -1 {
		return
	}
	endIndex := strings.Index(template, "]")
	if endIndex == -1 {
		return
	}
	format := template[startIndex+1 : endIndex]
	layout := DateFormatToLayout(format)
	c.filenameProvider = func(t time.Time) string {
		formatted := t.Format(layout)
		return strings.Replace(template, "["+format+"]", formatted, 1)
	}
}

//Validate valides configuration sttings
func (c *FileLoggerConfig) Validate() error {
	if len(c.LogType) == 0 {
		return errors.New("Log type was empty")
	}
	if c.FlushFrequencyInMs == 0 {
		c.FlushFrequencyInMs = c.FlushRequencyInMs
	}
	if c.FlushFrequencyInMs == 0 {
		return errors.New("FlushFrequencyInMs was 0")
	}

	if c.MaxQueueSize == 0 {
		return errors.New("MaxQueueSize was 0")
	}
	if len(c.FileTemplate) == 0 {
		return errors.New("FileTemplate was empty")
	}

	if c.MaxIddleTimeInSec == 0 {
		return errors.New("MaxIddleTimeInSec was 0")
	}
	if c.QueueFlashCount == 0 {
		return errors.New("QueueFlashCount was 0")
	}
	return nil
}

//LogStream represents individual log stream
type LogStream struct {
	Name             string
	Logger           *FileLogger
	Config           *FileLoggerConfig
	RecordCount      int
	File             *os.File
	LastAddQueueTime time.Time
	LastWriteTime    uint64
	Messages         chan string
	Complete         chan bool
}

//Log logs message into stream
func (s *LogStream) Log(message *LogMessage) error {
	if message == nil {
		return errors.New("message was nil")
	}
	var textMessage = ""
	var ok bool
	if textMessage, ok = message.Message.(string); ok {
	} else if IsStruct(message.Message) || IsMap(message.Message) || IsSlice(message.Message) {
		var buf = new(bytes.Buffer)
		err := NewJSONEncoderFactory().Create(buf).Encode(message.Message)
		if err != nil {
			return err
		}
		textMessage = strings.Trim(buf.String(), "\n\r")
	} else {
		return fmt.Errorf("unsupported type: %T", message.Message)
	}
	s.Messages <- textMessage
	s.LastAddQueueTime = time.Now()
	return nil
}

func (s *LogStream) write(message string) error {
	atomic.StoreUint64(&s.LastWriteTime, uint64(time.Now().UnixNano()))
	_, err := s.File.WriteString(message)
	if err != nil {
		return err
	}
	return s.File.Sync()
}

//Close closes stream.
func (s *LogStream) Close() {
	s.Logger.streamMapMutex.Lock()
	delete(s.Logger.streams, s.Name)
	s.Logger.streamMapMutex.Unlock()
	s.File.Close()

}

func (s *LogStream) isFrequencyFlushNeeded() bool {
	elapsedInMs := (int(time.Now().UnixNano()) - int(atomic.LoadUint64(&s.LastWriteTime))) / 1000000
	return elapsedInMs >= s.Config.FlushFrequencyInMs
}

func (s *LogStream) manageWritesInBatch() {
	messageCount := 0
	var message, messages string
	var timeout = time.Duration(2 * int(s.Config.FlushFrequencyInMs) * int(time.Millisecond))
	for {
		select {
		case done := <-s.Complete:
			if done {
				manageWritesInBatchLoopFlush(s, messageCount, messages)
				s.Close()
				os.Exit(0)
			}
		case <-time.After(timeout):
			if !manageWritesInBatchLoopFlush(s, messageCount, messages) {
				return
			}
			messageCount = 0
			messages = ""
		case message = <-s.Messages:
			messages += message + "\n"
			messageCount++
			s.RecordCount++

			var hasReachMaxRecrods = messageCount >= s.Config.QueueFlashCount && s.Config.QueueFlashCount > 0
			if hasReachMaxRecrods || s.isFrequencyFlushNeeded() {
				_ = s.write(messages)
				messages = ""
				messageCount = 0
			}

		}
	}
}

func manageWritesInBatchLoopFlush(s *LogStream, messageCount int, messages string) bool {
	if messageCount > 0 {
		if s.isFrequencyFlushNeeded() {
			err := s.write(messages)
			if err != nil {
				fmt.Printf("failed to write to log due to %v", err)
			}
			return true
		}
	}
	elapsedInMs := (int(time.Now().UnixNano()) - int(atomic.LoadUint64(&s.LastWriteTime))) / 1000000
	if elapsedInMs > s.Config.MaxIddleTimeInSec*1000 {
		s.Close()
		return false
	}
	return true
}

//FileLogger represents a file logger
type FileLogger struct {
	config         map[string]*FileLoggerConfig
	streamMapMutex *sync.RWMutex
	streams        map[string]*LogStream
	siginal        chan os.Signal
}

func (l *FileLogger) getConfig(messageType string) (*FileLoggerConfig, error) {
	config, found := l.config[messageType]
	if !found {
		return nil, errors.New("failed to lookup config for " + messageType)
	}
	config.Init()
	return config, nil
}

//NewLogStream creat a new LogStream for passed om path and file config
func (l *FileLogger) NewLogStream(path string, config *FileLoggerConfig) (*LogStream, error) {
	osFile, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644)
	if err != nil {
		return nil, err
	}
	logStream := &LogStream{Name: path, Logger: l, Config: config, File: osFile, Messages: make(chan string, config.MaxQueueSize), Complete: make(chan bool)}
	go func() {
		logStream.manageWritesInBatch()
	}()
	return logStream, nil
}

func (l *FileLogger) acquireLogStream(messageType string) (*LogStream, error) {
	config, err := l.getConfig(messageType)
	if err != nil {
		return nil, err
	}
	fileName := config.filenameProvider(time.Now())
	l.streamMapMutex.RLock()
	logStream, found := l.streams[fileName]
	l.streamMapMutex.RUnlock()
	if found {
		return logStream, nil
	}

	logStream, err = l.NewLogStream(fileName, config)
	if err != nil {
		return nil, err
	}
	l.streamMapMutex.Lock()
	l.streams[fileName] = logStream
	l.streamMapMutex.Unlock()
	return logStream, nil
}

//Log logs message into stream
func (l *FileLogger) Log(message *LogMessage) error {
	logStream, err := l.acquireLogStream(message.MessageType)
	if err != nil {
		return err
	}
	return logStream.Log(message)
}

//Notify notifies logger
func (l *FileLogger) Notify(siginal os.Signal) {
	l.siginal <- siginal
}

//NewFileLogger create new file logger
func NewFileLogger(configs ...FileLoggerConfig) (*FileLogger, error) {
	result := &FileLogger{
		config:         make(map[string]*FileLoggerConfig),
		streamMapMutex: &sync.RWMutex{},
		streams:        make(map[string]*LogStream),
	}

	for i := range configs {
		err := configs[i].Validate()
		if err != nil {
			return nil, err
		}
		result.config[configs[i].LogType] = &configs[i]
	}

	// If there's a signal to quit the program send it to channel
	result.siginal = make(chan os.Signal, 1)
	signal.Notify(result.siginal,
		syscall.SIGINT,
		syscall.SIGTERM)

	go func() {

		// Block until receive a quit signal
		_quit := <-result.siginal
		_ = _quit // don't care which type
		for _, stream := range result.streams {
			// No wait flush
			stream.Config.FlushFrequencyInMs = 0
			// Write logs now
			stream.Complete <- true
		}
	}()

	return result, nil
}