add haproxy json parser

main.go

@@ -199,8 +199,7 @@ func main() {
 	stopChan := make(chan bool)
 	stopHandlers := sync.WaitGroup{}
 
-	signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
-	signal.Notify(sigChan, os.Interrupt, syscall.SIGINT)
+	signal.Notify(sigChan, syscall.SIGHUP)
 
 	go func() {
 		sig := <-sigChan
@@ -210,13 +209,18 @@ func main() {
 		close(stopChan)
 		stopHandlers.Wait()
 
-		os.Exit(0)
+		switch sig {
+		case syscall.SIGHUP:
+			os.Exit(111)
+		default:
+			os.Exit(0)
+		}
 	}()
 
-	defer func() {
-		close(stopChan)
-		stopHandlers.Wait()
-	}()
+	//defer func() {
+	//close(stopChan)
+	//stopHandlers.Wait()
+	//}()
 
 	prof.SetupCPUProfiling(opts.CPUProfile, stopChan, &stopHandlers)
 	prof.SetupMemoryProfiling(opts.MemProfile, stopChan, &stopHandlers)
@@ -393,6 +397,7 @@ func processSource(nsCfg config.NamespaceConfig, t tail.Follower, parser parser.
 		}
 
 		fields, err := parser.ParseString(line)
+		fmt.Println(fields)
 		if err != nil {
 			fmt.Printf("error while parsing line '%s': %s\n", line, err)
 			metrics.parseErrorsTotal.Inc()

parser/haproxyjsonparser/haproxyjsonparser.go

@@ -0,0 +1,107 @@
+package haproxyjsonparser
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"regexp"
+	"strconv"
+	"strings"
+)
+
+// JsonParser parse a JSON string.
+type HaproxyJsonParser struct {
+	regexPrefix *regexp.Regexp
+}
+
+// NewJsonParser returns a new json parser.
+func NewHaproxyJsonParser() *HaproxyJsonParser {
+	regexPrefix := regexp.MustCompile("^.+{")
+	return &HaproxyJsonParser{
+		regexPrefix: regexPrefix,
+	}
+}
+
+// ParseString implements the Parser interface.
+// The value in the map is not necessarily a string, so it needs to be converted.
+func (j *HaproxyJsonParser) ParseString(line string) (map[string]string, error) {
+	result := map[string]string{
+		"request_length": "",
+		"time_local":     "",
+	}
+
+	line = j.regexPrefix.ReplaceAllString(line, "{")
+	parsed, err := j.parseJSON(line)
+	if err != nil {
+		return nil, err
+	}
+
+	// request
+	if v, ok := parsed["header"]; ok {
+		result["request"] = v
+	} else {
+		return nil, errors.New("Error parse request")
+	}
+
+	// status code
+	if v, ok := parsed["status"]; ok {
+		result["status"] = v
+	} else {
+		return nil, errors.New("Error parse status code")
+	}
+
+	// body bytes sent
+	if v, ok := parsed["bytes"]; ok {
+		result["body_bytes_sent"] = v
+	} else {
+		return nil, errors.New("Error parse body bytes sent")
+	}
+
+	// request method
+	splitRequestStr := strings.Split(result["request"], " ")
+	if len(splitRequestStr) == 0 {
+		return nil, errors.New("Error parse request method")
+	}
+	result["request_method"] = splitRequestStr[0]
+
+	// request_time
+	if v, ok := parsed["TR/Tw/Tc/Tr/Ta"]; ok {
+		timeSplit := strings.Split(v, "/")
+		if len(timeSplit) != 5 {
+			return nil, errors.New("Error parse request time")
+		}
+		intUpstreamTime, err := strconv.Atoi(timeSplit[3])
+		if err != nil {
+			return nil, errors.New("Error parse request time")
+		}
+		intRequestTime, err := strconv.Atoi(timeSplit[4])
+		if err != nil {
+			return nil, errors.New("Error parse request time")
+		}
+		// convert to second format
+		result["upstream_response_time"] = fmt.Sprintf("%.3f", float64(intUpstreamTime)/1000)
+		result["request_time"] = fmt.Sprintf("%.3f", float64(intRequestTime)/1000)
+	} else {
+		return nil, errors.New("Error parse request time")
+	}
+
+	return result, nil
+}
+
+func (j *HaproxyJsonParser) parseJSON(line string) (map[string]string, error) {
+	var parsed map[string]interface{}
+	err := json.Unmarshal([]byte(line), &parsed)
+	if err != nil {
+		return nil, fmt.Errorf("json log parsing err: %w", err)
+	}
+
+	fields := make(map[string]string, len(parsed))
+	for k, v := range parsed {
+		if s, ok := v.(string); ok {
+			fields[k] = s
+		} else {
+			fields[k] = fmt.Sprintf("%v", v)
+		}
+	}
+	return fields, nil
+}

parser/haproxyjsonparser/haproxyjsonparser_test.go

@@ -0,0 +1,30 @@
+package haproxyjsonparser
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestJsonParse(t *testing.T) {
+	parser := NewHaproxyJsonParser()
+	line := `Nov 16 05:01:27 somehost haproxy[24334]: {"time":"16/Nov/2021:05:01:27.794", "client":"192.168.10.3:17334", "frontend":"some-frontend", "backend":"backend/some-backend", "path":"/api/v1/users", "status":200, "bytes":872, "TR/Tw/Tc/Tr/Ta":"0/0/3/31/34", "CC":"-", "CS":"-", "tsc":"----", "concurrent_connections": "7/6/0/0/0", "queues":"0/0", "header":"POST /api/v1/users HTTP/1.1"}`
+
+	got, err := parser.ParseString(line)
+	require.NoError(t, err)
+
+	want := map[string]string{
+		"time_local":             "",
+		"request_time":           "0.034",
+		"request_length":         "",
+		"upstream_response_time": "0.031",
+		"status":                 "200",
+		"body_bytes_sent":        "872",
+		"request":                "POST /api/v1/users HTTP/1.1",
+		"request_method":         "POST",
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("JsonParser.Parse() = %v, want %v", got, want)
+	}
+}

parser/parser.go

@@ -2,6 +2,7 @@ package parser
 
 import (
 	"github.com/martin-helmich/prometheus-nginxlog-exporter/config"
+	"github.com/martin-helmich/prometheus-nginxlog-exporter/parser/haproxyjsonparser"
 	"github.com/martin-helmich/prometheus-nginxlog-exporter/parser/istioparser"
 	"github.com/martin-helmich/prometheus-nginxlog-exporter/parser/jsonparser"
 	"github.com/martin-helmich/prometheus-nginxlog-exporter/parser/kubeparser"
@@ -18,6 +19,8 @@ func NewParser(nsCfg config.NamespaceConfig) Parser {
 	switch nsCfg.Parser {
 	case "text":
 		return textparser.NewTextParser(nsCfg.Format)
+	case "haproxy-json":
+		return haproxyjsonparser.NewHaproxyJsonParser()
 	case "json":
 		return jsonparser.NewJsonParser()
 	case "kube":