log/influx.go at master · fanap-infra/log · GitHub

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
package log

import (
	"bytes"
	"fmt"
	"strings"
	"sync"
	"time"

	influxdb "github.com/influxdata/influxdb-client-go/v2"
	"github.com/influxdata/influxdb-client-go/v2/api"
)

const measurement = "logs"

type Influx struct {
	pool     sync.Pool
	client   influxdb.Client
	writeAPI api.WriteAPI
	app      string
}

func InfluxWriter(serverURL string, authToken string, org string, bucket string, app string, caller bool, stack EnablerFunc, enabler EnablerFunc) *Writer {
	i := &Influx{
		pool: sync.Pool{New: func() interface{} {
			b := bytes.NewBuffer(make([]byte, 150)) // buffer init with 150 size
			b.Reset()
			return b
		}},
		app: app,
	}
	i.Connect(serverURL, authToken, org, bucket)
	return newWriter(enabler, stack, caller, i)
}

func (i *Influx) Connect(serverURL string, authToken string, org string, bucket string) {
	i.client = influxdb.NewClient(serverURL, authToken)
	i.writeAPI = i.client.WriteAPI(org, bucket) // https://docs.influxdata.com/influxdb/v2.0/write-data/
}

func (i *Influx) close() {
	// Force all unwritten data to be sent
	i.writeAPI.Flush()
	// Ensures background processes finishes
	i.client.Close()
}

func (i *Influx) getBuffer() *bytes.Buffer {
	return i.pool.Get().(*bytes.Buffer)
}

func (i *Influx) putBuffer(b *bytes.Buffer) {
	b.Reset()
	i.pool.Put(b)
}

func (i *Influx) Print(l Level, s string, caller string, stack []string, message string) {
	fields := make(map[string]interface{})

	fields["message"] = message
	if caller != "" {
		fields["caller"] = caller
	}

	if len(stack) > 0 {
		fields["stack"] = strings.Join(stack, "\r\n")
	}

	// create point
	p := influxdb.NewPoint(
		measurement,
		map[string]string{
			"app":   i.app,
			"scope": s,
			"level": levelText[l],
		},
		fields,
		time.Now())

	// write asynchronously
	i.writeAPI.WritePoint(p)
}

func (i *Influx) Printv(l Level, s string, caller string, stack []string, message string, keysValues []interface{}) {
	fields := make(map[string]interface{})

	fields["message"] = message
	if caller != "" {
		fields["caller"] = caller
	}

	if len(stack) > 0 {
		fields["stack"] = strings.Join(stack, "\r\n")
	}

	i.addKeyValues(fields, keysValues)

	// create point
	p := influxdb.NewPoint(
		measurement,
		map[string]string{
			"app":   i.app,
			"scope": s,
			"level": levelText[l],
		},
		fields,
		time.Now())

	// write asynchronously
	i.writeAPI.WritePoint(p)
}

func (i *Influx) addKeyValues(fields map[string]interface{}, keysValues []interface{}) {
	lenValues := len(keysValues)
	if lenValues > 1 {
		values := i.getBuffer()
		defer i.putBuffer(values)

		// first key=value
		values.WriteString(fmt.Sprintf("%v=%v", keysValues[0], keysValues[1]))
		for i := 2; i < lenValues; i += 2 {
			values.WriteString(fmt.Sprintf("\r\n%v=%v", keysValues[i], keysValues[i+1]))
		}

		fields["values"] = values.Bytes()
	}
}