Merge pull request #19 from npande/npande_lambda_telemetry_api

npande · web-flow · commit 7d924b18995e · 2022-11-10T09:01:05.000+05:30
Checkin for lambda telemetry api support for extension.
diff --git a/README.md b/README.md
@@ -85,7 +85,7 @@ Releasing new layer versions
      `sh zip.sh`
 
 
-- The new wheel package gets released automatically after the tags are pushed using Github actions(Refer tagged-release in https://github.com/marvinpinto/action-automatic-releases).
+- The new extension binary and zip files gets released automatically after the tags are pushed using Github actions(Refer tagged-release in https://github.com/marvinpinto/action-automatic-releases).
 
    Run below commands to create and push tags
 
diff --git a/lambda-extensions/config/config.go b/lambda-extensions/config/config.go
@@ -35,6 +35,7 @@ type LambdaExtensionConfig struct {
 	LambdaRegion           string
 	SourceCategoryOverride string
 	EnhanceJsonLogs        bool
+	EnableSpanDrops        bool
 }
 
 var defaultLogTypes = []string{"platform", "function"}
@@ -66,6 +67,7 @@ func GetConfig() (*LambdaExtensionConfig, error) {
 	}
 	return config, nil
 }
+
 func (cfg *LambdaExtensionConfig) setDefaults() {
 	numRetry := os.Getenv("SUMO_NUM_RETRIES")
 	retrySleepTime := os.Getenv("SUMO_RETRY_SLEEP_TIME_MS")
@@ -75,6 +77,7 @@ func (cfg *LambdaExtensionConfig) setDefaults() {
 	enableFailover := os.Getenv("SUMO_ENABLE_FAILOVER")
 	logTypes := os.Getenv("SUMO_LOG_TYPES")
 	enhanceJsonLogs := os.Getenv("SUMO_ENHANCE_JSON_LOGS")
+	enableSpanDrops := os.Getenv("SUMO_SPAN_DROP")
 
 	if numRetry == "" {
 		cfg.NumRetry = 3
@@ -88,7 +91,6 @@ func (cfg *LambdaExtensionConfig) setDefaults() {
 	if maxConcurrentRequests == "" {
 		cfg.MaxConcurrentRequests = 3
 	}
-
 	if enableFailover == "" {
 		cfg.EnableFailover = false
 	}
@@ -106,6 +108,10 @@ func (cfg *LambdaExtensionConfig) setDefaults() {
 	if enhanceJsonLogs == "" {
 		cfg.EnhanceJsonLogs = true
 	}
+	if enableSpanDrops == "" {
+		// by default, spans will not be dropped if user did not configure the env variable
+		cfg.EnableSpanDrops = false
+	}
 }
 
 func (cfg *LambdaExtensionConfig) validateConfig() error {
@@ -116,6 +122,7 @@ func (cfg *LambdaExtensionConfig) validateConfig() error {
 	enableFailover := os.Getenv("SUMO_ENABLE_FAILOVER")
 	retrySleepTime := os.Getenv("SUMO_RETRY_SLEEP_TIME_MS")
 	enhanceJsonLogs := os.Getenv("SUMO_ENHANCE_JSON_LOGS")
+	enableSpanDrops := os.Getenv("SUMO_SPAN_DROP")
 
 	var allErrors []string
 	var err error
@@ -139,7 +146,7 @@ func (cfg *LambdaExtensionConfig) validateConfig() error {
 		}
 	}
 
-	if cfg.EnableFailover == true {
+	if cfg.EnableFailover {
 		if cfg.S3BucketName == "" {
 			allErrors = append(allErrors, "SUMO_S3_BUCKET_NAME not set in environment variable")
 		}
@@ -173,25 +180,24 @@ func (cfg *LambdaExtensionConfig) validateConfig() error {
 		} else {
 			cfg.MaxDataQueueLength = int(customMaxDataQueueLength)
 		}
-
 	}
+
 	if maxConcurrentRequests != "" {
 		customMaxConcurrentRequests, err := strconv.ParseInt(maxConcurrentRequests, 10, 32)
 		if err != nil {
 			allErrors = append(allErrors, fmt.Sprintf("Unable to parse SUMO_MAX_CONCURRENT_REQUESTS: %v", err))
 		} else {
 			cfg.MaxConcurrentRequests = int(customMaxConcurrentRequests)
 		}
-
 	}
+
 	if logLevel != "" {
 		customloglevel, err := logrus.ParseLevel(logLevel)
 		if err != nil {
 			allErrors = append(allErrors, fmt.Sprintf("Unable to parse SUMO_LOG_LEVEL: %v", err))
 		} else {
 			cfg.LogLevel = customloglevel
 		}
-
 	}
 
 	if enhanceJsonLogs != "" {
@@ -201,6 +207,13 @@ func (cfg *LambdaExtensionConfig) validateConfig() error {
 		}
 	}
 
+	if enableSpanDrops != "" {
+		cfg.EnableSpanDrops, err = strconv.ParseBool(enableSpanDrops)
+		if err != nil {
+			allErrors = append(allErrors, fmt.Sprintf("Unable to parse SUMO_SPAN_DROP: %v", err))
+		}
+	}
+
 	// test valid log format type
 	for _, logType := range cfg.LogTypes {
 		if !utils.StringInSlice(strings.TrimSpace(logType), validLogTypes) {
diff --git a/lambda-extensions/config/version.go b/lambda-extensions/config/version.go
@@ -8,7 +8,7 @@ import (
 
 // ExtensionName same as binary name or file name where main exists
 var ExtensionName = filepath.Base(os.Args[0])
-var layerVersion = "4"
+var layerVersion = "7"
 
 // SumoLogicExtensionLayerVersionSuffix denotes the layer version published in AWS
 var SumoLogicExtensionLayerVersionSuffix string = fmt.Sprintf("%s-prod:%s", ExtensionName, layerVersion)
diff --git a/lambda-extensions/lambdaapi/telemetryapiclient.go b/lambda-extensions/lambdaapi/telemetryapiclient.go
@@ -0,0 +1,47 @@
+package lambdaapi
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+)
+
+const (
+	// Base URL for telemetry api extension
+	telemetryURL = "2022-07-01/telemetry"
+	// Subscription Body Constants. Subscribe to platform logs and receive them on ${local_ip}:4243 via HTTP protocol.
+	telemetry_timeoutMs    = 1000
+	telemetry_maxBytes     = 1048576
+	telemetry_maxItems     = 10000
+	telemetry_receiverPort = 4243
+)
+
+// SubscribeToLogsAPI is - Subscribe to Logs API to receive the Lambda Logs.
+func (client *Client) SubscribeToTelemetryAPI(ctx context.Context, logEvents []string) ([]byte, error) {
+	URL := client.baseURL + telemetryURL
+
+	reqBody, err := json.Marshal(map[string]interface{}{
+		"destination":   map[string]interface{}{"protocol": "HTTP", "URI": fmt.Sprintf("http://sandbox:%v", receiverPort)},
+		"types":         logEvents,
+		"buffering":     map[string]interface{}{"timeoutMs": telemetry_timeoutMs, "maxBytes": telemetry_maxBytes, "maxItems": telemetry_maxItems},
+		"schemaVersion": "2022-07-01",
+	})
+	if err != nil {
+		return nil, err
+	}
+	headers := map[string]string{
+		extensionIdentiferHeader: client.extensionID,
+	}
+	var response []byte
+	if ctx != nil {
+		response, err = client.MakeRequestWithContext(ctx, headers, bytes.NewBuffer(reqBody), "PUT", URL)
+	} else {
+		response, err = client.MakeRequest(headers, bytes.NewBuffer(reqBody), "PUT", URL)
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	return response, nil
+}
diff --git a/lambda-extensions/lambdaapi/telemetryapiclient_test.go b/lambda-extensions/lambdaapi/telemetryapiclient_test.go
@@ -0,0 +1,35 @@
+package lambdaapi
+
+import (
+	"context"
+	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestSubscribeToTelemetryAPI(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assertEqual(t, r.Method, http.MethodPut, "Method is not PUT")
+		assertNotEmpty(t, r.Header.Get(extensionNameHeader), "Extension Name Header not present")
+
+		reqBytes, err := ioutil.ReadAll(r.Body)
+		assertNoError(t, err, "Received error")
+		defer r.Body.Close()
+		assertNotEmpty(t, reqBytes, "Received error in request")
+
+		w.Header().Add(extensionIdentiferHeader, "test-sumo-id")
+		w.WriteHeader(200)
+	}))
+
+	defer srv.Close()
+	client := NewClient(srv.URL[7:], extensionName)
+
+	// Without Context
+	response, err := client.SubscribeToTelemetryAPI(nil, []string{"platform", "function", "extension"})
+	commonAsserts(t, client, response, err)
+
+	// With Context
+	response, err = client.SubscribeToTelemetryAPI(context.Background(), []string{"platform", "function", "extension"})
+	commonAsserts(t, client, response, err)
+}
diff --git a/lambda-extensions/sumoclient/sumoclient.go b/lambda-extensions/sumoclient/sumoclient.go
@@ -101,7 +101,7 @@ func (s *sumoLogicClient) failoverHandler(buf *bytes.Buffer) error {
 		}
 		err = utils.UploadToS3(&s.config.S3BucketName, &keyName, buf)
 		if err != nil {
-			err = fmt.Errorf("Failed to Send to S3 Bucket %s Path %s: %w", s.config.S3BucketName, keyName, err)
+			err = fmt.Errorf("failed to send to s3 bucket %s path %s: %w", s.config.S3BucketName, keyName, err)
 		}
 		return err
 	}
@@ -159,13 +159,22 @@ func (s *sumoLogicClient) createCWLogLine(item map[string]interface{}) {
 
 	message, ok := item["record"].(map[string]interface{})
 	if ok {
-		delete(item, "record")
+		s.logger.Debug("Not dropping record, if logType is platform.report.")
+		// delete(item, "record")
 	}
+
 	// Todo convert this to struct
+	// Updated cwMessageLine to also cover new field initDurationMs as record.metrics do have it.
 	metric := message["metrics"].(map[string]interface{})
-	cwMessageLine := fmt.Sprintf("REPORT RequestId: %v	Duration: %v ms	Billed Duration: %v ms 	Memory Size: %v MB	Max Memory Used: %v MB",
-		message["requestId"], metric["durationMs"], metric["billedDurationMs"], metric["memorySizeMB"], metric["maxMemoryUsedMB"])
-	item["message"] = cwMessageLine
+	if metric["initDurationMs"] == nil {
+		cwMessageLine := fmt.Sprintf("REPORT RequestId: %v	Duration: %v ms	Billed Duration: %v ms 	Memory Size: %v MB	Max Memory Used: %v MB",
+			message["requestId"], metric["durationMs"], metric["billedDurationMs"], metric["memorySizeMB"], metric["maxMemoryUsedMB"])
+		item["message"] = cwMessageLine
+	} else {
+		cwMessageLine := fmt.Sprintf("REPORT RequestId: %v	Duration: %v ms	Billed Duration: %v ms 	Memory Size: %v MB	Max Memory Used: %v MB	Init Duration: %v ms",
+			message["requestId"], metric["durationMs"], metric["billedDurationMs"], metric["memorySizeMB"], metric["maxMemoryUsedMB"], metric["initDurationMs"])
+		item["message"] = cwMessageLine
+	}
 }
 
 func (s *sumoLogicClient) getLogGroup() string {
@@ -210,17 +219,26 @@ func (s *sumoLogicClient) enhanceLogs(msg responseBody) {
 			}
 		} else if ok && logType == "platform.report" {
 			s.createCWLogLine(item)
+		} else if ok && logType == "platform.runtimeDone" {
+			message, ok := item["record"].(map[string]interface{})
+			if ok {
+				_, ok := message["spans"]
+				if ok && s.config.EnableSpanDrops {
+					// dropping spans if its present and configured to drop
+					delete(message, "spans")
+				}
+			}
 		}
 	}
 }
 
 func (s *sumoLogicClient) transformBytesToArrayOfMap(rawmsg []byte) (responseBody, error) {
 	s.logger.Debugln("Transforming bytes to array of maps")
 	var msg responseBody
-	var err error
-	err = json.Unmarshal(rawmsg, &msg)
+	// var err error
+	var err error = json.Unmarshal(rawmsg, &msg)
 	if err != nil {
-		return msg, fmt.Errorf("Error in parsing payload %s: %v", string(rawmsg), err)
+		return msg, fmt.Errorf("error in parsing payload %s: %v", string(rawmsg), err)
 	}
 	return msg, err
 }
@@ -253,7 +271,7 @@ func (s *sumoLogicClient) createChunks(msgArr responseBody) ([]string, error) {
 	}
 	chunks = append(chunks, currentChunk.String())
 	if errorCount > 0 {
-		err = fmt.Errorf("Dropping %d messages due to json parsing error", errorCount)
+		err = fmt.Errorf("dropping %d messages due to json parsing error", errorCount)
 	}
 	s.logger.Debugf("Chunks created: %d NumOfParsingError: %d", len(chunks), errorCount)
 	return chunks, err
diff --git a/lambda-extensions/sumologic-extension.go b/lambda-extensions/sumologic-extension.go
@@ -64,13 +64,14 @@ func runTimeAPIInit() (int64, error) {
 	}
 	logger.Debug("Succcessfully Registered with Run Time API Client: ", utils.PrettyPrint(registerResponse))
 
-	// Subscribe to Logs API
-	logger.Debug("Subscribing Extension to Logs API........")
-	subscribeResponse, err := extensionClient.SubscribeToLogsAPI(nil, config.LogTypes)
+	// Subscribe to Telemetry API
+	logger.Debug("Subscribing Extension to Telemetry API........")
+	subscribeResponse, err := extensionClient.SubscribeToTelemetryAPI(nil, config.LogTypes)
 	if err != nil {
 		return 0, err
 	}
-	logger.Debug("Successfully subscribed to Logs API: ", utils.PrettyPrint(string(subscribeResponse)))
+
+	logger.Debug("Successfully subscribed to Telemetry API: ", utils.PrettyPrint(string(subscribeResponse)))
 
 	// Call next to say registration is successful and get the deadtimems
 	nextResponse, err := nextEvent(nil)

Original file line number	Diff line number	Diff line change
`@@ -35,6 +35,7 @@ type LambdaExtensionConfig struct {`
`35`	`35`	`LambdaRegion string`
`36`	`36`	`SourceCategoryOverride string`
`37`	`37`	`EnhanceJsonLogs bool`
	`38`	`+ EnableSpanDrops bool`
`38`	`39`	`}`
`39`	`40`
`40`	`41`	`var defaultLogTypes = []string{"platform", "function"}`
`@@ -66,6 +67,7 @@ func GetConfig() (*LambdaExtensionConfig, error) {`
`66`	`67`	`}`
`67`	`68`	`return config, nil`
`68`	`69`	`}`
	`70`	`+`
`69`	`71`	`func (cfg *LambdaExtensionConfig) setDefaults() {`
`70`	`72`	`numRetry := os.Getenv("SUMO_NUM_RETRIES")`
`71`	`73`	`retrySleepTime := os.Getenv("SUMO_RETRY_SLEEP_TIME_MS")`
`@@ -75,6 +77,7 @@ func (cfg *LambdaExtensionConfig) setDefaults() {`
`75`	`77`	`enableFailover := os.Getenv("SUMO_ENABLE_FAILOVER")`
`76`	`78`	`logTypes := os.Getenv("SUMO_LOG_TYPES")`
`77`	`79`	`enhanceJsonLogs := os.Getenv("SUMO_ENHANCE_JSON_LOGS")`
	`80`	`+ enableSpanDrops := os.Getenv("SUMO_SPAN_DROP")`
`78`	`81`
`79`	`82`	`if numRetry == "" {`
`80`	`83`	`cfg.NumRetry = 3`
`@@ -88,7 +91,6 @@ func (cfg *LambdaExtensionConfig) setDefaults() {`
`88`	`91`	`if maxConcurrentRequests == "" {`
`89`	`92`	`cfg.MaxConcurrentRequests = 3`
`90`	`93`	`}`
`91`		`-`
`92`	`94`	`if enableFailover == "" {`
`93`	`95`	`cfg.EnableFailover = false`
`94`	`96`	`}`
`@@ -106,6 +108,10 @@ func (cfg *LambdaExtensionConfig) setDefaults() {`
`106`	`108`	`if enhanceJsonLogs == "" {`
`107`	`109`	`cfg.EnhanceJsonLogs = true`
`108`	`110`	`}`
	`111`	`+ if enableSpanDrops == "" {`
	`112`	`+ // by default, spans will not be dropped if user did not configure the env variable`
	`113`	`+ cfg.EnableSpanDrops = false`
	`114`	`+ }`
`109`	`115`	`}`
`110`	`116`
`111`	`117`	`func (cfg *LambdaExtensionConfig) validateConfig() error {`
`@@ -116,6 +122,7 @@ func (cfg *LambdaExtensionConfig) validateConfig() error {`
`116`	`122`	`enableFailover := os.Getenv("SUMO_ENABLE_FAILOVER")`
`117`	`123`	`retrySleepTime := os.Getenv("SUMO_RETRY_SLEEP_TIME_MS")`
`118`	`124`	`enhanceJsonLogs := os.Getenv("SUMO_ENHANCE_JSON_LOGS")`
	`125`	`+ enableSpanDrops := os.Getenv("SUMO_SPAN_DROP")`
`119`	`126`
`120`	`127`	`var allErrors []string`
`121`	`128`	`var err error`
`@@ -139,7 +146,7 @@ func (cfg *LambdaExtensionConfig) validateConfig() error {`
`139`	`146`	`}`
`140`	`147`	`}`
`141`	`148`
`142`		`- if cfg.EnableFailover == true {`
	`149`	`+ if cfg.EnableFailover {`
`143`	`150`	`if cfg.S3BucketName == "" {`
`144`	`151`	`allErrors = append(allErrors, "SUMO_S3_BUCKET_NAME not set in environment variable")`
`145`	`152`	`}`
`@@ -173,25 +180,24 @@ func (cfg *LambdaExtensionConfig) validateConfig() error {`
`173`	`180`	`} else {`
`174`	`181`	`cfg.MaxDataQueueLength = int(customMaxDataQueueLength)`
`175`	`182`	`}`
`176`		`-`
`177`	`183`	`}`
	`184`	`+`
`178`	`185`	`if maxConcurrentRequests != "" {`
`179`	`186`	`customMaxConcurrentRequests, err := strconv.ParseInt(maxConcurrentRequests, 10, 32)`
`180`	`187`	`if err != nil {`
`181`	`188`	`allErrors = append(allErrors, fmt.Sprintf("Unable to parse SUMO_MAX_CONCURRENT_REQUESTS: %v", err))`
`182`	`189`	`} else {`
`183`	`190`	`cfg.MaxConcurrentRequests = int(customMaxConcurrentRequests)`
`184`	`191`	`}`
`185`		`-`
`186`	`192`	`}`
	`193`	`+`
`187`	`194`	`if logLevel != "" {`
`188`	`195`	`customloglevel, err := logrus.ParseLevel(logLevel)`
`189`	`196`	`if err != nil {`
`190`	`197`	`allErrors = append(allErrors, fmt.Sprintf("Unable to parse SUMO_LOG_LEVEL: %v", err))`
`191`	`198`	`} else {`
`192`	`199`	`cfg.LogLevel = customloglevel`
`193`	`200`	`}`
`194`		`-`
`195`	`201`	`}`
`196`	`202`
`197`	`203`	`if enhanceJsonLogs != "" {`
`@@ -201,6 +207,13 @@ func (cfg *LambdaExtensionConfig) validateConfig() error {`
`201`	`207`	`}`
`202`	`208`	`}`
`203`	`209`
	`210`	`+ if enableSpanDrops != "" {`
	`211`	`+ cfg.EnableSpanDrops, err = strconv.ParseBool(enableSpanDrops)`
	`212`	`+ if err != nil {`
	`213`	`+ allErrors = append(allErrors, fmt.Sprintf("Unable to parse SUMO_SPAN_DROP: %v", err))`
	`214`	`+ }`
	`215`	`+ }`
	`216`	`+`
`204`	`217`	`// test valid log format type`
`205`	`218`	`for _, logType := range cfg.LogTypes {`
`206`	`219`	`if !utils.StringInSlice(strings.TrimSpace(logType), validLogTypes) {`
Original file line number	Diff line number	Diff line change
`@@ -101,7 +101,7 @@ func (s sumoLogicClient) failoverHandler(buf bytes.Buffer) error {`
`101`	`101`	`}`
`102`	`102`	`err = utils.UploadToS3(&s.config.S3BucketName, &keyName, buf)`
`103`	`103`	`if err != nil {`
`104`		`- err = fmt.Errorf("Failed to Send to S3 Bucket %s Path %s: %w", s.config.S3BucketName, keyName, err)`
	`104`	`+ err = fmt.Errorf("failed to send to s3 bucket %s path %s: %w", s.config.S3BucketName, keyName, err)`
`105`	`105`	`}`
`106`	`106`	`return err`
`107`	`107`	`}`
`@@ -159,13 +159,22 @@ func (s *sumoLogicClient) createCWLogLine(item map[string]interface{}) {`
`159`	`159`
`160`	`160`	`message, ok := item["record"].(map[string]interface{})`
`161`	`161`	`if ok {`
`162`		`- delete(item, "record")`
	`162`	`+ s.logger.Debug("Not dropping record, if logType is platform.report.")`
	`163`	`+ // delete(item, "record")`
`163`	`164`	`}`
	`165`	`+`
`164`	`166`	`// Todo convert this to struct`
	`167`	`+ // Updated cwMessageLine to also cover new field initDurationMs as record.metrics do have it.`
`165`	`168`	`metric := message["metrics"].(map[string]interface{})`
`166`		`- cwMessageLine := fmt.Sprintf("REPORT RequestId: %v Duration: %v ms Billed Duration: %v ms Memory Size: %v MB Max Memory Used: %v MB",`
`167`		`- message["requestId"], metric["durationMs"], metric["billedDurationMs"], metric["memorySizeMB"], metric["maxMemoryUsedMB"])`
`168`		`- item["message"] = cwMessageLine`
	`169`	`+ if metric["initDurationMs"] == nil {`
	`170`	`+ cwMessageLine := fmt.Sprintf("REPORT RequestId: %v Duration: %v ms Billed Duration: %v ms Memory Size: %v MB Max Memory Used: %v MB",`
	`171`	`+ message["requestId"], metric["durationMs"], metric["billedDurationMs"], metric["memorySizeMB"], metric["maxMemoryUsedMB"])`
	`172`	`+ item["message"] = cwMessageLine`
	`173`	`+ } else {`
	`174`	`+ cwMessageLine := fmt.Sprintf("REPORT RequestId: %v Duration: %v ms Billed Duration: %v ms Memory Size: %v MB Max Memory Used: %v MB Init Duration: %v ms",`
	`175`	`+ message["requestId"], metric["durationMs"], metric["billedDurationMs"], metric["memorySizeMB"], metric["maxMemoryUsedMB"], metric["initDurationMs"])`
	`176`	`+ item["message"] = cwMessageLine`
	`177`	`+ }`
`169`	`178`	`}`
`170`	`179`
`171`	`180`	`func (s *sumoLogicClient) getLogGroup() string {`
`@@ -210,17 +219,26 @@ func (s *sumoLogicClient) enhanceLogs(msg responseBody) {`
`210`	`219`	`}`
`211`	`220`	`} else if ok && logType == "platform.report" {`
`212`	`221`	`s.createCWLogLine(item)`
	`222`	`+ } else if ok && logType == "platform.runtimeDone" {`
	`223`	`+ message, ok := item["record"].(map[string]interface{})`
	`224`	`+ if ok {`
	`225`	`+ _, ok := message["spans"]`
	`226`	`+ if ok && s.config.EnableSpanDrops {`
	`227`	`+ // dropping spans if its present and configured to drop`
	`228`	`+ delete(message, "spans")`
	`229`	`+ }`
	`230`	`+ }`
`213`	`231`	`}`
`214`	`232`	`}`
`215`	`233`	`}`
`216`	`234`
`217`	`235`	`func (s *sumoLogicClient) transformBytesToArrayOfMap(rawmsg []byte) (responseBody, error) {`
`218`	`236`	`s.logger.Debugln("Transforming bytes to array of maps")`
`219`	`237`	`var msg responseBody`
`220`		`- var err error`
`221`		`- err = json.Unmarshal(rawmsg, &msg)`
	`238`	`+ // var err error`
	`239`	`+ var err error = json.Unmarshal(rawmsg, &msg)`
`222`	`240`	`if err != nil {`
`223`		`- return msg, fmt.Errorf("Error in parsing payload %s: %v", string(rawmsg), err)`
	`241`	`+ return msg, fmt.Errorf("error in parsing payload %s: %v", string(rawmsg), err)`
`224`	`242`	`}`
`225`	`243`	`return msg, err`
`226`	`244`	`}`
`@@ -253,7 +271,7 @@ func (s *sumoLogicClient) createChunks(msgArr responseBody) ([]string, error) {`
`253`	`271`	`}`
`254`	`272`	`chunks = append(chunks, currentChunk.String())`
`255`	`273`	`if errorCount > 0 {`
`256`		`- err = fmt.Errorf("Dropping %d messages due to json parsing error", errorCount)`
	`274`	`+ err = fmt.Errorf("dropping %d messages due to json parsing error", errorCount)`
`257`	`275`	`}`
`258`	`276`	`s.logger.Debugf("Chunks created: %d NumOfParsingError: %d", len(chunks), errorCount)`
`259`	`277`	`return chunks, err`