Historical processing only targets last year

config/config.go

@@ -25,12 +25,13 @@ type MonitorConfig struct {
 
 // SourceConfig holds the config that defines all data sources to be processed.
 type SourceConfig struct {
-	Bucket     string   `yaml:"bucket"`
-	Experiment string   `yaml:"experiment"`
-	Datatype   string   `yaml:"datatype"`
-	Filter     string   `yaml:"filter"`
-	Datasets   Datasets `yaml:"target_datasets"`
-	DailyOnly  bool     `yaml:"daily_only"`
+	Bucket      string   `yaml:"bucket"`
+	Experiment  string   `yaml:"experiment"`
+	Datatype    string   `yaml:"datatype"`
+	Filter      string   `yaml:"filter"`
+	Datasets    Datasets `yaml:"target_datasets"`
+	DailyOnly   bool     `yaml:"daily_only"`
+	FullHistory bool     `yaml:"full_history"`
 }
 
 // Datasets contains the name of BigQuery datasets used for temporary, raw, or

config/config.yml

@@ -27,6 +27,7 @@ sources:
     tmp: tmp_ndt
     raw: raw_ndt
     join: ndt
+  full_history: true
 - bucket: archive-{{NDT_SOURCE_PROJECT}}
   experiment: ndt
   datatype: pcap

job-service/iterators.go

@@ -168,6 +168,12 @@ func (j *JobIterator) Next() (*tracker.JobWithTarget, error) {
 	// Copy the jobspec and set the date.
 	jt := &tracker.JobWithTarget{}
 	*jt = j.specs[j.nextIndex]
+
+	lastYear := time.Now().UTC().AddDate(-1, 0, 0)
+	if !jt.FullHistory && j.lastDate.Before(lastYear) {
+		return nil, nil
+	}
+
 	jt.Job.Date = j.lastDate
 	jt.ID = jt.Job.Key()
 	j.nextIndex++ // increment index for next call.

job-service/job-service.go

@@ -56,6 +56,12 @@ func (svc *Service) NextJob(ctx context.Context) *tracker.JobWithTarget {
 			log.Println(err)
 			continue
 		}
+
+		lastYear := time.Now().UTC().AddDate(-1, 0, 0)
+		if !jt.FullHistory && jt.Job.Date.Before(lastYear) {
+			return nil
+		}
+
 		return svc.ifHasFiles(ctx, jt)
 	}
 	return nil
@@ -108,8 +114,9 @@ func NewJobService(startDate time.Time,
 		// TODO - handle gs:// targets
 		jt := tracker.JobWithTarget{
 			// NOTE: JobWithTarget.ID is assigned after Job.Date is set.
-			Job:       job,
-			DailyOnly: s.DailyOnly,
+			Job:         job,
+			DailyOnly:   s.DailyOnly,
+			FullHistory: s.FullHistory,
 		}
 		dailySpecs = append(dailySpecs, jt)
 		if !s.DailyOnly {

tracker/job.go

@@ -52,9 +52,10 @@ func (j *Job) TablePartition() string {
 // JobWithTarget specifies a type/date job, and a destination
 // table or GCS prefix
 type JobWithTarget struct {
-	ID        Key // ID used by gardener & parsers to identify a Job's status and configuration.
-	Job       Job
-	DailyOnly bool `json:"-"`
+	ID          Key // ID used by gardener & parsers to identify a Job's status and configuration.
+	Job         Job
+	DailyOnly   bool `json:"-"`
+	FullHistory bool `json:"-"`
 	// TODO: enable configuration for parser to target alterate buckets.
 }