Merge pull request #114 from kube-logging/framing

loggen: add octet-counting framing functionality

Author: pepov

README.md

@@ -75,8 +75,9 @@ curl --location --request POST 'localhost:11000/loggen' \
 --header 'Content-Type: application/json' \
 --data-raw '{
     "type": "web",
-    "format": "nginx.access",
-    "count": 1000
+    "format": "nginx",
+    "count": 1000,
+    "framing": false
 }'
 ```
 
@@ -85,8 +86,9 @@ Response:
 ```json
 {
   "type": "web",
-  "format": "nginx.access",
-  "count": 1000
+  "format": "nginx",
+  "count": 1000,
+  "framing": false
 }
 ```
 

conf/config.ini.sample

@@ -29,3 +29,7 @@
 
 #[nginx]
 #enabled = true
+
+#[destination]
+#network = "tcp"
+#address = "127.0.0.1:514"

formats/golang/golang.go

@@ -42,6 +42,8 @@ type GolangLog struct {
 	Level       string `json:"level"`
 	MSG         string `json:"msg"`
 	Time        string `json:"time"`
+
+	isFramed bool
 }
 
 func (g GolangLog) newRandomMessage() string {
@@ -65,7 +67,7 @@ func (g GolangLog) newRandomMessage() string {
 	return msgList[g.Level]
 }
 
-func NewGolangLogRandom(i GolangLogIntensity) GolangLog {
+func NewGolangLogRandom(i GolangLogIntensity) *GolangLog {
 	rand.Seed(time.Now().UTC().UnixNano())
 	c, err := wr.NewChooser(
 		wr.Choice{Item: "error", Weight: *i.ErrorWeight},
@@ -76,7 +78,7 @@ func NewGolangLogRandom(i GolangLogIntensity) GolangLog {
 	if err != nil {
 		log.Error(err)
 	}
-	return GolangLog{
+	return &GolangLog{
 		Application: randomdata.StringSample("webshop", "blog"),
 		Environment: randomdata.StringSample("production", "sandbox", "demo"),
 		Component:   randomdata.StringSample("frontend", "backend", "worker"),
@@ -104,6 +106,14 @@ func (g GolangLog) String() (string, float64) {
 	return message, float64(len([]byte(message)))
 }
 
+func (l *GolangLog) IsFramed() bool {
+	return l.isFramed
+}
+
+func (l *GolangLog) SetFramed(f bool) {
+	l.isFramed = f
+}
+
 func (g GolangLog) Labels() prometheus.Labels {
 	return prometheus.Labels{
 		"type":     "golang",

log/log.go

@@ -21,4 +21,6 @@ import (
 type Log interface {
 	String() (string, float64)
 	Labels() prometheus.Labels
+	IsFramed() bool
+	SetFramed(bool)
 }

log/template.go

@@ -34,6 +34,7 @@ type LogTemplate struct {
 
 	template *template.Template
 	data     LogTemplateData
+	isFramed bool
 }
 
 func NewLogTemplate(format string, fs fs.FS, data LogTemplateData) (*LogTemplate, error) {
@@ -70,6 +71,14 @@ func (l *LogTemplate) String() (string, float64) {
 	return str, float64(len([]byte(str)))
 }
 
+func (l *LogTemplate) IsFramed() bool {
+	return l.isFramed
+}
+
+func (l *LogTemplate) SetFramed(f bool) {
+	l.isFramed = f
+}
+
 func (l *LogTemplate) Labels() prometheus.Labels {
 	return prometheus.Labels{
 		"type":     l.Format,

loggen/loggen.go

@@ -52,9 +52,10 @@ type LogGen struct {
 }
 
 type LogGenRequest struct {
-	Type   string `json:"type"`
-	Format string `json:"format"`
-	Count  int    `json:"count"`
+	Type    string `json:"type"`
+	Format  string `json:"format"`
+	Count   int    `json:"count"`
+	Framing bool   `json:"framing"`
 }
 
 func New() *LogGen {
@@ -214,6 +215,10 @@ func (l *LogGen) processRequests() bool {
 			continue
 		}
 
+		if request.Framing {
+			msg.SetFramed(true)
+		}
+
 		logs = append(logs, msg)
 		e = e.Next()
 	}

loggen/writers.go

@@ -40,6 +40,11 @@ func newStdoutWriter() LogWriter {
 
 func (*StdoutLogWriter) Send(l log.Log) {
 	msg, size := l.String()
+
+	if l.IsFramed() {
+		msg = fmt.Sprintf("%d %s", len(msg), msg)
+	}
+
 	fmt.Println(msg)
 
 	metrics.EventEmitted.With(l.Labels()).Inc()
@@ -67,7 +72,12 @@ func newNetworkWriter(network string, address string) LogWriter {
 func (nlw *NetworkLogWriter) Send(l log.Log) {
 	msg, size := l.String()
 
-	msg += "\n"
+	if l.IsFramed() {
+		msg = fmt.Sprintf("%d %s", len(msg), msg)
+	} else {
+		msg += "\n"
+	}
+
 	written := 0
 	for {
 		data := msg[written:]