Merge pull request #17 from SumoLogic/hpal_lambda_cost_fixes

Fixes Lambda Extension cost overruns

Author: himanshu219

lambda-extensions/config/config.go

@@ -28,7 +28,6 @@ type LambdaExtensionConfig struct {
 	LogLevel               logrus.Level
 	MaxDataQueueLength     int
 	MaxConcurrentRequests  int
-	ProcessingSleepTime    time.Duration
 	MaxRetryAttempts       int
 	RetrySleepTime         time.Duration
 	ConnectionTimeoutValue time.Duration
@@ -37,7 +36,8 @@ type LambdaExtensionConfig struct {
 	SourceCategoryOverride string
 }
 
-var validLogTypes = []string{"platform", "function"}
+var defaultLogTypes = []string{"platform", "function"}
+var validLogTypes = []string{"platform", "function", "extension"}
 
 // GetConfig to get config instance
 func GetConfig() (*LambdaExtensionConfig, error) {
@@ -52,7 +52,6 @@ func GetConfig() (*LambdaExtensionConfig, error) {
 		LambdaRegion:           os.Getenv("AWS_REGION"),
 		SourceCategoryOverride: os.Getenv("SOURCE_CATEGORY_OVERRIDE"),
 		MaxRetryAttempts:       5,
-		RetrySleepTime:         300 * time.Millisecond,
 		ConnectionTimeoutValue: 10000 * time.Millisecond,
 		MaxDataPayloadSize:     1024 * 1024, // 1 MB
 	}
@@ -68,12 +67,13 @@ func GetConfig() (*LambdaExtensionConfig, error) {
 }
 func (cfg *LambdaExtensionConfig) setDefaults() {
 	numRetry := os.Getenv("SUMO_NUM_RETRIES")
-	processingSleepTime := os.Getenv("SUMO_PROCESSING_SLEEP_TIME_MS")
+	retrySleepTime := os.Getenv("SUMO_RETRY_SLEEP_TIME_MS")
 	logLevel := os.Getenv("SUMO_LOG_LEVEL")
 	maxDataQueueLength := os.Getenv("SUMO_MAX_DATAQUEUE_LENGTH")
 	maxConcurrentRequests := os.Getenv("SUMO_MAX_CONCURRENT_REQUESTS")
 	enableFailover := os.Getenv("SUMO_ENABLE_FAILOVER")
 	logTypes := os.Getenv("SUMO_LOG_TYPES")
+
 	if numRetry == "" {
 		cfg.NumRetry = 3
 	}
@@ -94,12 +94,12 @@ func (cfg *LambdaExtensionConfig) setDefaults() {
 		cfg.AWSLambdaRuntimeAPI = "127.0.0.1:9001"
 	}
 	if logTypes == "" {
-		cfg.LogTypes = validLogTypes
+		cfg.LogTypes = defaultLogTypes
 	} else {
 		cfg.LogTypes = strings.Split(logTypes, ",")
 	}
-	if processingSleepTime == "" {
-		cfg.ProcessingSleepTime = 0 * time.Millisecond
+	if retrySleepTime == "" {
+		cfg.RetrySleepTime =  300 * time.Millisecond
 	}
 
 }
@@ -110,7 +110,7 @@ func (cfg *LambdaExtensionConfig) validateConfig() error {
 	maxDataQueueLength := os.Getenv("SUMO_MAX_DATAQUEUE_LENGTH")
 	maxConcurrentRequests := os.Getenv("SUMO_MAX_CONCURRENT_REQUESTS")
 	enableFailover := os.Getenv("SUMO_ENABLE_FAILOVER")
-	processingSleepTime := os.Getenv("SUMO_PROCESSING_SLEEP_TIME_MS")
+	retrySleepTime := os.Getenv("SUMO_RETRY_SLEEP_TIME_MS")
 
 	var allErrors []string
 	var err error
@@ -152,12 +152,12 @@ func (cfg *LambdaExtensionConfig) validateConfig() error {
 		}
 	}
 
-	if processingSleepTime != "" {
-		customProcessingSleepTime, err := strconv.ParseInt(processingSleepTime, 10, 32)
+	if retrySleepTime != "" {
+		customRetrySleepTime, err := strconv.ParseInt(retrySleepTime, 10, 32)
 		if err != nil {
-			allErrors = append(allErrors, fmt.Sprintf("Unable to parse SUMO_PROCESSING_SLEEP_TIME_MS: %v", err))
+			allErrors = append(allErrors, fmt.Sprintf("Unable to parse SUMO_RETRY_SLEEP_TIME_MS: %v", err))
 		} else {
-			cfg.ProcessingSleepTime = time.Duration(customProcessingSleepTime) * time.Millisecond
+			cfg.RetrySleepTime = time.Duration(customRetrySleepTime) * time.Millisecond
 		}
 	}
 

lambda-extensions/sumoclient/sumoclient_test.go

@@ -31,6 +31,8 @@ func setupEnv() {
 	os.Setenv("SUMO_MAX_DATAQUEUE_LENGTH", "10")
 	os.Setenv("SUMO_MAX_CONCURRENT_REQUESTS", "3")
 	os.Setenv("SUMO_LOG_LEVEL", "DEBUG")
+	os.Setenv("SUMO_RETRY_SLEEP_TIME_MS", "50")
+	os.Setenv("SUMO_LOG_TYPES", "function")
 }
 
 func assertEqual(t *testing.T, a interface{}, b interface{}, message string) {

lambda-extensions/sumologic-extension.go

@@ -104,18 +104,16 @@ func processEvents(ctx context.Context) {
 			consumer.FlushDataQueue(ctx)
 			return
 		default:
+			logger.Debugf("switching to other go routine")
+			runtime.Gosched()
 			logger.Infof("Calling DrainQueue from processEvents")
-			for {
-				runtime_done := consumer.DrainQueue(ctx)
-
-				if runtime_done == 1 {
-					logger.Infof("Exiting DrainQueueLoop: Runtime is Done")
-					break
-				} else {
-					logger.Debugf("switching to other go routine")
-					runtime.Gosched()
-				}
+			// for {
+			runtime_done := consumer.DrainQueue(ctx)
+
+			if runtime_done == 1 {
+				logger.Infof("Exiting DrainQueueLoop: Runtime is Done")
 			}
+			// }
 
 			// This statement will freeze lambda
 			nextResponse, err := nextEvent(ctx)