GROUNDWORK-1690 Organize data processing

Author: Pavlo Sumkin

connectors/checker-connector/main.go

@@ -98,31 +98,31 @@ func taskHandler(task ScheduleTask) func() {
 		cmd := exec.Command(task.Command[0], task.Command[1:]...)
 		cmd.Env = task.Environment
 		var (
-			handler     func() ([]byte, error)
-			err         error
-			res         []byte
-			span, spanN services.TraceSpan
-			ctx, ctxN   context.Context
+			handler func() ([]byte, error)
+			err     error
+			res     []byte
+
+			monitoredResources *[]transit.DynamicMonitoredResource
 		)
 		if task.CombinedOutput {
 			handler = cmd.CombinedOutput
 		} else {
 			handler = cmd.Output
 		}
 
-		ctx, span = services.StartTraceSpan(context.Background(), "connectors", "taskHandler")
+		ctx, span := services.StartTraceSpan(context.Background(), "connectors", "taskHandler")
 		defer func() {
 			services.EndTraceSpan(span,
 				services.TraceAttrError(err),
 				services.TraceAttrPayloadLen(res),
 				services.TraceAttrString("task", task.String()),
 			)
 		}()
-		_, spanN = services.StartTraceSpan(ctx, "connectors", "command")
 
+		_, span2 := services.StartTraceSpan(ctx, "connectors", "command")
 		res, err = handler()
 
-		services.EndTraceSpan(spanN,
+		services.EndTraceSpan(span2,
 			services.TraceAttrError(err),
 			services.TraceAttrPayloadLen(res),
 			services.TraceAttrArray("command", task.Command),
@@ -140,18 +140,26 @@ func taskHandler(task ScheduleTask) func() {
 			Bytes("res", res).
 			Msg("task done")
 
-		ctxN, spanN = services.StartTraceSpan(ctx, "connectors", "processMetrics")
+		_, span3 := services.StartTraceSpan(ctx, "connectors", "parse")
+		monitoredResources, err = parser.Parse(res, task.DataFormat)
+
+		services.EndTraceSpan(span3,
+			services.TraceAttrError(err),
+			services.TraceAttrPayloadLen(res),
+		)
 
-		if _, err = parser.ProcessMetrics(ctxN, res, task.DataFormat); err != nil {
+		if err != nil {
 			log.Warn().Err(err).
 				Interface("task", task).
 				Bytes("res", res).
-				Msg("could not process metrics")
+				Msg("could not parse metrics")
+			return
+		}
+		if err = connectors.SendMetrics(ctx, *monitoredResources, nil); err != nil {
+			log.Warn().Err(err).
+				Interface("task", task).
+				Bytes("res", res).
+				Msg("could not send metrics")
 		}
-
-		services.EndTraceSpan(spanN,
-			services.TraceAttrError(err),
-			services.TraceAttrPayloadLen(res),
-		)
 	}
 }

connectors/nsca-connector/main.go

@@ -40,7 +40,7 @@ func main() {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	nscaCancel = cancel
-	nsca.Start(ctx)
+	nsca.Start(ctx, makeNSCAHandler())
 
 	/* return on quit signal */
 	<-transitService.Quit()

connectors/nsca-connector/nsca/nsca.go

@@ -13,53 +13,54 @@ import (
 	"strings"
 	"time"
 
-	"github.com/gwos/tcg/connectors/nsca-connector/parser"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 	"github.com/tubemogul/nscatools"
 )
 
-func Handler(p *DataPacketExt) error {
+func AdaptHandler(h func([]byte) error) DataHandler {
 	// It's unclear for now how to process multi-line metrics in the right way.
 	// The nscatools.DataPacket provides host, service, and state info for the 1st line only.
 	// And PluginOutput contains the rest.
 	// The `fullPacket` payload that processed in the HandleClient() by nscatools.DataPacket.Read()
 	// contains a lot of noise and looks like protocol defined structure.
 	// So, here we try to reconstruct plain metrics payload and process it with own parser.
-	s := strings.Replace(p.PluginOutput, `\n`, "\n", -1)
-	buf := make([]byte, 0, 4+len(p.HostName)+len(p.Service)+len(s))
-	buf = append(buf, p.HostName...)
-	buf = append(buf, ';')
-	buf = append(buf, p.Service...)
-	buf = append(buf, ';')
-	buf = strconv.AppendInt(buf, int64(p.State), 10)
-	buf = append(buf, ';')
-	buf = append(buf, s...)
-	log.Debug().
-		Int16("version", p.Version).
-		Uint32("crc", p.Crc).
-		Uint32("timestamp", p.Timestamp).
-		Int16("state", p.State).
-		Str("hostname", p.HostName).
-		Str("service", p.Service).
-		Str("pluginOutput", p.PluginOutput).
-		Bytes("buf", buf).
-		Func(func(e *zerolog.Event) {
-			println("# NSCA plain packet #")
-			println(string(buf))
-		}).
-		Msg("processing DataPacket")
-
-	_, err := parser.ProcessMetrics(context.Background(), buf, parser.NSCA)
-	if err != nil {
-		log.Warn().Err(err).
+	return func(p *DataPacketExt) error {
+		s := strings.Replace(p.PluginOutput, `\n`, "\n", -1)
+		buf := make([]byte, 0, 4+len(p.HostName)+len(p.Service)+len(s))
+		buf = append(buf, p.HostName...)
+		buf = append(buf, ';')
+		buf = append(buf, p.Service...)
+		buf = append(buf, ';')
+		buf = strconv.AppendInt(buf, int64(p.State), 10)
+		buf = append(buf, ';')
+		buf = append(buf, s...)
+		log.Debug().
+			Int16("version", p.Version).
+			Uint32("crc", p.Crc).
+			Uint32("timestamp", p.Timestamp).
+			Int16("state", p.State).
+			Str("hostname", p.HostName).
+			Str("service", p.Service).
+			Str("pluginOutput", p.PluginOutput).
 			Bytes("buf", buf).
-			Msg("could not process metrics")
+			Func(func(e *zerolog.Event) {
+				println("# NSCA plain packet #")
+				println(string(buf))
+			}).
+			Msg("processing DataPacket")
+
+		err := h(buf)
+		if err != nil {
+			log.Warn().Err(err).
+				Bytes("payload", buf).
+				Msg("could not process incoming data")
+		}
+		return nil
 	}
-	return err
 }
 
-func Start(ctx context.Context) {
+func Start(ctx context.Context, handler DataHandler) {
 	nscaHost := "0.0.0.0"
 	nscaPort := uint16(5667)
 	nscaEncrypt := nscatools.EncryptNone
@@ -86,7 +87,7 @@ func Start(ctx context.Context) {
 	}
 
 	go StartServerWithContext(ctx,
-		NewConfigExt(nscaHost, nscaPort, nscaEncrypt, nscaPassword, Handler))
+		NewConfigExt(nscaHost, nscaPort, nscaEncrypt, nscaPassword, handler))
 }
 
 func StartServerWithContext(ctx context.Context, conf *ConfigExt) error {
@@ -156,14 +157,14 @@ func HandleClientExt(conf *ConfigExt, conn net.Conn) error {
 	return err
 }
 
-type dataHandler func(*DataPacketExt) error
+type DataHandler func(*DataPacketExt) error
 
 type ConfigExt struct {
 	nscatools.Config
-	PacketHandler dataHandler
+	PacketHandler DataHandler
 }
 
-func NewConfigExt(host string, port uint16, encryption int, password string, handler dataHandler) *ConfigExt {
+func NewConfigExt(host string, port uint16, encryption int, password string, handler DataHandler) *ConfigExt {
 	c := nscatools.NewConfig(host, port, encryption, password, nil)
 	cfg := ConfigExt{*c, handler}
 	return &cfg

connectors/nsca-connector/nscaConnector.go

@@ -6,6 +6,8 @@ import (
 	"net/http"
 
 	"github.com/gin-gonic/gin"
+	"github.com/gwos/tcg/connectors"
+	"github.com/gwos/tcg/connectors/nsca-connector/nsca"
 	"github.com/gwos/tcg/connectors/nsca-connector/parser"
 	"github.com/gwos/tcg/services"
 	"github.com/rs/zerolog/log"
@@ -46,7 +48,7 @@ func makeEntrypointHandler(dataFormat parser.DataFormat) func(*gin.Context) {
 			c.JSON(http.StatusBadRequest, err.Error())
 			return
 		}
-		if _, err = parser.ProcessMetrics(ctx, payload, dataFormat); err != nil {
+		if err = processData(ctx, payload, dataFormat); err != nil {
 			log.Warn().Err(err).
 				Str("entrypoint", c.FullPath()).
 				Str("dataFormat", string(dataFormat)).
@@ -58,3 +60,36 @@ func makeEntrypointHandler(dataFormat parser.DataFormat) func(*gin.Context) {
 		c.JSON(http.StatusOK, nil)
 	}
 }
+
+func makeNSCAHandler() nsca.DataHandler {
+	return nsca.AdaptHandler(func(p []byte) error {
+		ctx, span := services.StartTraceSpan(context.Background(), "connectors", "EntrypointHandler")
+		err := processData(ctx, p, parser.NSCA)
+		if err != nil {
+			log.Warn().Err(err).
+				Str("entrypoint", "NSCA").
+				Msg("could not process incoming request")
+		}
+		services.EndTraceSpan(span,
+			services.TraceAttrError(err),
+			services.TraceAttrPayloadLen(p),
+			services.TraceAttrEntrypoint("NSCA"),
+		)
+		return err
+	})
+}
+
+func processData(ctx context.Context, payload []byte, dataFormat parser.DataFormat) error {
+	ctxN, span := services.StartTraceSpan(ctx, "connectors", "processData")
+	monitoredResources, err := parser.Parse(payload, dataFormat)
+
+	services.EndTraceSpan(span,
+		services.TraceAttrError(err),
+		services.TraceAttrPayloadLen(payload),
+	)
+
+	if err != nil {
+		return err
+	}
+	return connectors.SendMetrics(ctxN, *monitoredResources, nil)
+}

connectors/nsca-connector/parser/parser.go

@@ -2,7 +2,6 @@ package parser
 
 import (
 	"bytes"
-	"context"
 	"errors"
 	"fmt"
 	"regexp"
@@ -12,7 +11,6 @@ import (
 
 	"github.com/gwos/tcg/connectors"
 	"github.com/gwos/tcg/milliseconds"
-	"github.com/gwos/tcg/services"
 	"github.com/gwos/tcg/transit"
 )
 
@@ -42,32 +40,7 @@ const (
 type MetricsMap map[string][]transit.TimeSeries
 type ServicesMap map[string][]transit.DynamicMonitoredService
 
-func ProcessMetrics(ctx context.Context, payload []byte, dataFormat DataFormat) (*[]transit.DynamicMonitoredResource, error) {
-	var (
-		ctxN               context.Context
-		err                error
-		monitoredResources *[]transit.DynamicMonitoredResource
-		span               services.TraceSpan
-	)
-
-	ctxN, span = services.StartTraceSpan(ctx, "connectors", "parseBody")
-	monitoredResources, err = parse(payload, dataFormat)
-
-	services.EndTraceSpan(span,
-		services.TraceAttrError(err),
-		services.TraceAttrPayloadLen(payload),
-	)
-
-	if err != nil {
-		return nil, err
-	}
-	if err := connectors.SendMetrics(ctxN, *monitoredResources, nil); err != nil {
-		return nil, err
-	}
-	return monitoredResources, nil
-}
-
-func parse(payload []byte, dataFormat DataFormat) (*[]transit.DynamicMonitoredResource, error) {
+func Parse(payload []byte, dataFormat DataFormat) (*[]transit.DynamicMonitoredResource, error) {
 	metricsLines := strings.Split(string(bytes.Trim(payload, " \n\r")), "\n")
 
 	var (

connectors/nsca-connector/parser/parser_test.go

@@ -14,7 +14,7 @@ Server3;Disks3;2;CRITICAL - load average: 2.45, 2.32, 2.22|load1=2.450;0.000;0.0
 awips-demo-4;example-service-10;0;OK - example-service-10 (2021-08-09 15:47:38 :: 1628524058) | result=147ms;;;0;`,
 	)
 
-	monitoredResources, err := parse(data, NSCA)
+	monitoredResources, err := Parse(data, NSCA)
 	assert.NoError(t, err)
 
 	assert.Equal(t, 4, len(*monitoredResources), "invalid count of monitored resources")
@@ -56,7 +56,7 @@ S;1628530909;awips-demo-2;example-service-6;0;OK - example-service-6 (2021-08-09
 S;1628530909;awips-demo-2;example-service-7;0;OK - example-service-7 (2021-08-09 17:41:49 :: 1628530909) | result=63ms;;;0;`,
 	)
 
-	monitoredResources, err := parse(data, Bronx)
+	monitoredResources, err := Parse(data, Bronx)
 	assert.NoError(t, err)
 
 	assert.Equal(t, 4, len(*monitoredResources), "invalid count of monitored resources")
@@ -94,7 +94,7 @@ S;1628546296;awips-demo-2;example-service-4;0;OK - example-service-4 (2021-08-09
 S;1628546296;awips-demo-2;example-service-5;0;OK - example-service-5 (2021-08-09 21:58:16 :: 1628546296) | result=63ms;;;0;`,
 	)
 
-	monitoredResources, err := parse(data, Bronx)
+	monitoredResources, err := Parse(data, Bronx)
 	assert.NoError(t, err)
 
 	assert.Equal(t, 1, len(*monitoredResources), "invalid count of monitored resources")