configurable time extraction window

Author: rjtokenring

README.md

@@ -73,6 +73,7 @@ These parameters are filled by CFT at stack creation time and can be adjusted la
 | /arduino/s3-importer/iot/filter/tags    | (optional) tags filtering. Syntax: tag=value,tag2=value2  |
 | /arduino/s3-importer/iot/samples-resolution  | (optional) samples aggregation resolution (1/5/15 minutes, 1 hour, raw) |
 | /arduino/s3-importer/destination-bucket  | S3 destination bucket |
+| /arduino/s3-importer/iot/scheduling | Execution scheduling |
 
 ### Tag filtering
 

business/tsextractor/tsextractor.go

@@ -52,7 +52,8 @@ func (a *TsExtractor) ExportTSToS3(
 	resolution int,
 	destinationS3Bucket string) error {
 
-	to := time.Now().Truncate(time.Hour).UTC()
+	// Truncate time to given resolution
+	to := time.Now().Truncate(time.Duration(resolution) * time.Second).UTC()
 	from := to.Add(-time.Duration(timeWindowInMinutes) * time.Minute)
 
 	// Open s3 output writer

deployment/cloud-formation-template/deployment.yaml

@@ -171,6 +171,15 @@ Resources:
         Ref: DestinationS3Bucket
       Tier: Standard
 
+  ExecutionSchedulingParameter:
+    Type: AWS::SSM::Parameter
+    Properties:
+      Name: /arduino/s3-importer/iot/scheduling
+      Type: String
+      Value:
+        Ref: ExecutionScheduling
+      Tier: Standard
+
   # EventBridge Rule to trigger Lambda every hour
   EventBridgeRule:
     Type: AWS::Events::Rule

lambda.go

@@ -31,16 +31,17 @@ type AWSS3ImportTrigger struct {
 }
 
 const (
-	ArduinoPrefix               = "/arduino/s3-importer"
-	IoTApiKey                   = ArduinoPrefix + "/iot/api-key"
-	IoTApiSecret                = ArduinoPrefix + "/iot/api-secret"
-	IoTApiOrgId                 = ArduinoPrefix + "/iot/org-id"
-	IoTApiTags                  = ArduinoPrefix + "/iot/filter/tags"
-	SamplesResoSec              = ArduinoPrefix + "/iot/samples-resolution-seconds"
-	SamplesReso                 = ArduinoPrefix + "/iot/samples-resolution"
-	DestinationS3Bucket         = ArduinoPrefix + "/destination-bucket"
-	SamplesResolutionSeconds    = 300
-	TimeExtractionWindowMinutes = 60
+	ArduinoPrefix                      = "/arduino/s3-importer"
+	IoTApiKey                          = ArduinoPrefix + "/iot/api-key"
+	IoTApiSecret                       = ArduinoPrefix + "/iot/api-secret"
+	IoTApiOrgId                        = ArduinoPrefix + "/iot/org-id"
+	IoTApiTags                         = ArduinoPrefix + "/iot/filter/tags"
+	SamplesResoSec                     = ArduinoPrefix + "/iot/samples-resolution-seconds"
+	SamplesReso                        = ArduinoPrefix + "/iot/samples-resolution"
+	Scheduling                         = ArduinoPrefix + "/iot/scheduling"
+	DestinationS3Bucket                = ArduinoPrefix + "/destination-bucket"
+	SamplesResolutionSeconds           = 300
+	DefaultTimeExtractionWindowMinutes = 60
 )
 
 func HandleRequest(ctx context.Context, event *AWSS3ImportTrigger) (*string, error) {
@@ -78,6 +79,8 @@ func HandleRequest(ctx context.Context, event *AWSS3ImportTrigger) (*string, err
 	if tagsParam != nil {
 		tags = tagsParam
 	}
+
+	// Resolve resolution
 	resolution, err := paramReader.ReadIntConfig(SamplesResoSec)
 	if err != nil {
 		// Possibly this parameter is not set. Try SamplesReso
@@ -101,12 +104,29 @@ func HandleRequest(ctx context.Context, event *AWSS3ImportTrigger) (*string, err
 		}
 		resolution = &val
 	}
-
 	if *resolution > 3600 {
 		logger.Errorf("Resolution %d is invalid", *resolution)
 		return nil, errors.New("resolution must be between -1 and 3600")
 	}
 
+	// Resolve scheduling
+	schedule, err := paramReader.ReadConfig(Scheduling)
+	if err != nil {
+		logger.Error("Error reading parameter "+Scheduling, err)
+		return nil, err
+	}
+	extractionWindowMinutes := DefaultTimeExtractionWindowMinutes
+	switch *schedule {
+	case "5 minutes":
+		extractionWindowMinutes = 5
+	case "15 minutes":
+		extractionWindowMinutes = 15
+	case "1 hour":
+		extractionWindowMinutes = 60
+	case "1 day":
+		extractionWindowMinutes = 24 * 60
+	}
+
 	logger.Infoln("------ Running import...")
 	if event.Dev || os.Getenv("DEV") == "true" {
 		logger.Infoln("Running in dev mode")
@@ -127,8 +147,9 @@ func HandleRequest(ctx context.Context, event *AWSS3ImportTrigger) (*string, err
 	} else {
 		logger.Infoln("resolution:", *resolution, "seconds")
 	}
+	logger.Infoln("data extraction time windows:", extractionWindowMinutes, "minutes")
 
-	err = importer.StartImport(ctx, logger, *apikey, *apiSecret, organizationId, tags, *resolution, TimeExtractionWindowMinutes, *destinationS3Bucket)
+	err = importer.StartImport(ctx, logger, *apikey, *apiSecret, organizationId, tags, *resolution, extractionWindowMinutes, *destinationS3Bucket)
 	if err != nil {
 		return nil, err
 	}