added TELEMETRY_TIMEOUT_MS, TELEMETRY_MAX_BYTES and TELEMETRY_MAX_ITEMS in environment config

Author: himanshu219

lambda-extensions/config/config.go

@@ -38,6 +38,9 @@ type LambdaExtensionConfig struct {
 	EnhanceJsonLogs        bool
 	EnableSpanDrops        bool
 	KmsCacheSeconds        int64
+	TelemetryTimeoutMs     int
+	TelemetryMaxBytes      int64
+	TelemetryMaxItems      int
 }
 
 var defaultLogTypes = []string{"platform", "function"}
@@ -82,6 +85,22 @@ func (cfg *LambdaExtensionConfig) setDefaults() {
 	enhanceJsonLogs := os.Getenv("SUMO_ENHANCE_JSON_LOGS")
 	enableSpanDrops := os.Getenv("SUMO_SPAN_DROP")
 	kmsCacheSeconds := os.Getenv("KMS_CACHE_SECONDS")
+	telemetryTimeoutMs := os.Getenv("TELEMETRY_TIMEOUT_MS")
+	telemetryMaxBytes := os.Getenv("TELEMETRY_MAX_BYTES")
+	telemetryMaxItems := os.Getenv("TELEMETRY_MAX_ITEMS")
+
+	if telemetryTimeoutMs == "" {
+		cfg.TelemetryTimeoutMs = 1000
+	}
+
+	if telemetryMaxBytes == "" {
+		cfg.TelemetryMaxBytes = 262144
+	}
+
+	if telemetryMaxItems == "" {
+		cfg.TelemetryMaxItems = 10000
+	}
+
 
 	if numRetry == "" {
 		cfg.NumRetry = 3
@@ -141,6 +160,9 @@ func (cfg *LambdaExtensionConfig) validateConfig() error {
 	enhanceJsonLogs := os.Getenv("SUMO_ENHANCE_JSON_LOGS")
 	enableSpanDrops := os.Getenv("SUMO_SPAN_DROP")
 	kmsCacheSeconds := os.Getenv("KMS_CACHE_SECONDS")
+	telemetryTimeoutMs := os.Getenv("TELEMETRY_TIMEOUT_MS")
+	telemetryMaxBytes := os.Getenv("TELEMETRY_MAX_BYTES")
+	telemetryMaxItems := os.Getenv("TELEMETRY_MAX_ITEMS")
 
 	var allErrors []string
 	var err error
@@ -239,6 +261,38 @@ func (cfg *LambdaExtensionConfig) validateConfig() error {
 		}
 	}
 
+	if telemetryTimeoutMs != "" {
+		telemetryTimeoutMs, err := strconv.ParseInt(telemetryTimeoutMs, 10, 32)
+		if err != nil {
+			allErrors = append(allErrors, fmt.Sprintf("Unable to parse TELEMETRY_TIMEOUT_MS: %v", err))
+		} else {
+			cfg.TelemetryTimeoutMs = int(telemetryTimeoutMs)
+		}
+		cfg.TelemetryTimeoutMs = max(cfg.TelemetryTimeoutMs, 25)
+		cfg.TelemetryTimeoutMs = min(cfg.TelemetryTimeoutMs, 30000)
+	}
+
+	if telemetryMaxBytes != "" {
+		cfg.TelemetryMaxBytes, err = strconv.ParseInt(telemetryMaxBytes, 10, 64)
+		if err != nil {
+			allErrors = append(allErrors, fmt.Sprintf("Unable to parse TELEMETRY_MAX_BYTES: %v", err))
+		}
+		cfg.TelemetryMaxBytes = max(cfg.TelemetryMaxBytes, 262144)
+		cfg.TelemetryMaxBytes = min(cfg.TelemetryMaxBytes, 1048576)
+	}
+
+	if telemetryMaxItems != "" {
+		telemetryMaxItems, err := strconv.ParseInt(telemetryMaxItems, 10, 32)
+		if err != nil {
+			allErrors = append(allErrors, fmt.Sprintf("Unable to parse TELEMETRY_MAX_ITEMS: %v", err))
+		} else {
+			cfg.TelemetryMaxItems = int(telemetryMaxItems)
+		}
+		cfg.TelemetryMaxItems = max(cfg.TelemetryMaxItems, 1000)
+		cfg.TelemetryMaxItems = min(cfg.TelemetryMaxItems, 10000)
+	}
+
+
 	// test valid log format type
 	for _, logType := range cfg.LogTypes {
 		if !utils.StringInSlice(strings.TrimSpace(logType), validLogTypes) {

lambda-extensions/lambdaapi/telemetryapiclient.go

@@ -11,20 +11,17 @@ 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) {
+func (client *Client) SubscribeToTelemetryAPI(ctx context.Context, logEvents []string, telemetryTimeoutMs int, telemetryMaxBytes int64, telemetryMaxItems int) ([]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},
+		"buffering":     map[string]interface{}{"timeoutMs": telemetryTimeoutMs, "maxBytes": telemetryMaxBytes, "maxItems": telemetryMaxItems},
 		"schemaVersion": "2022-07-01",
 	})
 	if err != nil {

lambda-extensions/lambdaapi/telemetryapiclient_test.go

@@ -26,10 +26,10 @@ func TestSubscribeToTelemetryAPI(t *testing.T) {
 	client := NewClient(srv.URL[7:], extensionName)
 
 	// Without Context
-	response, err := client.SubscribeToTelemetryAPI(nil, []string{"platform", "function", "extension"})
+	response, err := client.SubscribeToTelemetryAPI(nil, []string{"platform", "function", "extension"}, 1000, 262144, 10000)
 	commonAsserts(t, client, response, err)
 
 	// With Context
-	response, err = client.SubscribeToTelemetryAPI(context.Background(), []string{"platform", "function", "extension"})
+	response, err = client.SubscribeToTelemetryAPI(context.Background(), []string{"platform", "function", "extension"}, 1000, 262144, 10000)
 	commonAsserts(t, client, response, err)
 }

lambda-extensions/sumologic-extension.go

@@ -66,7 +66,7 @@ func runTimeAPIInit() (int64, error) {
 
 	// Subscribe to Telemetry API
 	logger.Debug("Subscribing Extension to Telemetry API........")
-	subscribeResponse, err := extensionClient.SubscribeToTelemetryAPI(nil, config.LogTypes)
+	subscribeResponse, err := extensionClient.SubscribeToTelemetryAPI(nil, config.LogTypes, config.TelemetryTimeoutMs, config.TelemetryMaxBytes, config.TelemetryMaxItems)
 	if err != nil {
 		return 0, err
 	}