Merge pull request #5 from williamchanrico/support_time_offset

Add support for cron-job-time-offset flag

Author: williamchanrico

Makefile

@@ -3,7 +3,7 @@
 NAME          := planet-exporter
 BIN_DIRECTORY := ./bin
 REVISION      := $(shell git rev-parse --short HEAD 2>/dev/null)
-VERSION       := v0.1.0
+VERSION       := v0.1.1
 
 ifndef REVISION
 	override REVISION = none

cmd/planet-federator/internal/internal.go

@@ -32,11 +32,21 @@ import (
 // Config contains main service config options
 type Config struct {
 	// Main config
+	// CronJobSchedule schedule using cron format used by the Quartz Scheduler
+	// 1. Seconds
+	// 2. Minutes
+	// 3. Hours
+	// 4. Day-of-Month
+	// 5. Month
+	// 6. Day-of-Week
+	// 7. Year (optional field)
 	CronJobSchedule      string
 	CronJobTimeoutSecond int
-	LogLevel             string
-	LogDisableTimestamp  bool
-	LogDisableColors     bool
+	// CronJobTimeOffset all cron job start time (e.g. '-5m' will query data from 5 minutes ago)
+	CronJobTimeOffset   time.Duration
+	LogLevel            string
+	LogDisableTimestamp bool
+	LogDisableColors    bool
 
 	InfluxdbAddr      string
 	InfluxdbToken     string
@@ -118,16 +128,28 @@ func (s Service) Run(ctx context.Context) error {
 	return nil
 }
 
+// getCronJobStartTime returns the time for cron job starting point
+func (s Service) getCronJobStartTime() time.Time {
+	// We want to offset the query time by the specified offset
+	return time.Now().Add(s.Config.CronJobTimeOffset)
+}
+
+// getCronJobDuration returns the duration since the cron job was started
+func (s Service) getCronJobDuration(startTime time.Time) time.Duration {
+	// We want to offset the query time by the specified offset
+	return time.Now().Add(s.Config.CronJobTimeOffset).Sub(startTime)
+}
+
 // TrafficBandwidthJobFunc queries traffic bandwidth (planet-exporter) data from Prometheus and store
 // them in federator backend
 func (s Service) TrafficBandwidthJobFunc() {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(s.Config.CronJobTimeoutSecond)*time.Second)
 	defer cancel()
 
-	jobStartTime := time.Now()
+	jobStartTime := s.getCronJobStartTime()
 	log.Debugf("A job started: %v", jobStartTime)
 
-	trafficPeers, err := s.PrometheusSvc.QueryPlanetExporterTrafficBandwidth(ctx, time.Now().Add(-15*time.Second), time.Now())
+	trafficPeers, err := s.PrometheusSvc.QueryPlanetExporterTrafficBandwidth(ctx, jobStartTime.Add(-15*time.Second), jobStartTime)
 	if err != nil {
 		log.Errorf("Error querying traffic peers from prometheus: %v", err)
 	}
@@ -140,10 +162,10 @@ func (s Service) TrafficBandwidthJobFunc() {
 			RemoteDomain:    trafficPeer.RemoteDomain,
 			BitsPerSecond:   trafficPeer.BandwidthBitsPerSecond,
 			Direction:       trafficPeer.Direction,
-		}, time.Now())
+		}, jobStartTime)
 	}
 
-	log.Infof("Traffic Bandwidth Job took: %v", time.Since(jobStartTime))
+	log.Infof("Traffic Bandwidth Job took: %v", s.getCronJobDuration(jobStartTime))
 }
 
 // UpstreamServicesJobFunc queries upstream services (planet-exporter) data from Prometheus and store
@@ -152,10 +174,10 @@ func (s Service) UpstreamServicesJobFunc() {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(s.Config.CronJobTimeoutSecond)*time.Second)
 	defer cancel()
 
-	jobStartTime := time.Now()
+	jobStartTime := s.getCronJobStartTime()
 	log.Debugf("A job started: %v", jobStartTime)
 
-	upstreamServices, err := s.PrometheusSvc.QueryPlanetExporterUpstreamServices(ctx, time.Now().Add(-15*time.Second), time.Now())
+	upstreamServices, err := s.PrometheusSvc.QueryPlanetExporterUpstreamServices(ctx, jobStartTime.Add(-15*time.Second), jobStartTime)
 	if err != nil {
 		log.Errorf("Error querying upstream services from prometheus: %v", err)
 	}
@@ -169,10 +191,10 @@ func (s Service) UpstreamServicesJobFunc() {
 			UpstreamHostgroup: svc.UpstreamHostgroup,
 			UpstreamAddress:   svc.UpstreamAddress,
 			Protocol:          svc.Protocol,
-		}, time.Now())
+		}, jobStartTime)
 	}
 
-	log.Infof("Upstream Service Job took: %v", time.Since(jobStartTime))
+	log.Infof("Upstream Service Job took: %v", s.getCronJobDuration(jobStartTime))
 }
 
 // DownstreamServicesJobFunc queries downstream services (planet-exporter) data from Prometheus and store
@@ -181,10 +203,10 @@ func (s Service) DownstreamServicesJobFunc() {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(s.Config.CronJobTimeoutSecond)*time.Second)
 	defer cancel()
 
-	jobStartTime := time.Now()
+	jobStartTime := s.getCronJobStartTime()
 	log.Debugf("A job started: %v", jobStartTime)
 
-	downstreamServices, err := s.PrometheusSvc.QueryPlanetExporterDownstreamServices(ctx, time.Now().Add(-15*time.Second), time.Now())
+	downstreamServices, err := s.PrometheusSvc.QueryPlanetExporterDownstreamServices(ctx, jobStartTime.Add(-15*time.Second), jobStartTime)
 	if err != nil {
 		log.Errorf("Error querying downstream services from prometheus: %v", err)
 	}
@@ -198,8 +220,8 @@ func (s Service) DownstreamServicesJobFunc() {
 			DownstreamHostgroup: svc.DownstreamHostgroup,
 			DownstreamAddress:   svc.DownstreamAddress,
 			Protocol:            svc.Protocol,
-		}, time.Now())
+		}, jobStartTime)
 	}
 
-	log.Infof("Downstream Service Job took: %v", time.Since(jobStartTime))
+	log.Infof("Downstream Service Job took: %v", s.getCronJobDuration(jobStartTime))
 }

cmd/planet-federator/main.go

@@ -19,6 +19,7 @@ import (
 	"flag"
 	"fmt"
 	"os"
+	"time"
 
 	"planet-exporter/cmd/planet-federator/internal"
 	federator "planet-exporter/federator"
@@ -37,11 +38,20 @@ var (
 )
 
 func main() {
+	var err error
 	var config internal.Config
 
+	// cronJobTimeOffsetDuration allows federator to go back in time. For example,
+	// set '-10h30m' to tell federator to offset query time to 10 hours 30 minutes ago.
+	//
+	// This is useful when we want to integrate federator to existing Prometheus setup.
+	// TODO: Allows running multiple jobs for federator to catch up faster.
+	var cronJobTimeOffsetDuration string
+
 	// Main
-	flag.StringVar(&config.CronJobSchedule, "cron-job-schedule", "*/30 * * * * *", "Cron jobs schedule to pre-process planet-exporter metrics into federator backend")
+	flag.StringVar(&config.CronJobSchedule, "cron-job-schedule", "*/30 * * * * *", "Cron jobs schedule (Quartz Scheduler format: s m h dom mo dow y) to pre-process planet-exporter metrics into federator backend")
 	flag.IntVar(&config.CronJobTimeoutSecond, "cron-job-timeout-second", 30, "Timeout per federator job in second")
+	flag.StringVar(&cronJobTimeOffsetDuration, "cron-job-time-offset", "0s", "Cron jobs time offset. (e.g. '-1h5m' to query data from 1 hour 5 minutes ago)")
 	flag.StringVar(&config.LogLevel, "log-level", "info", "Log level")
 	flag.BoolVar(&config.LogDisableTimestamp, "log-disable-timestamp", false, "Disable timestamp on logger")
 	flag.BoolVar(&config.LogDisableColors, "log-disable-colors", false, "Disable colors on logger")
@@ -64,6 +74,11 @@ func main() {
 		os.Exit(0)
 	}
 
+	config.CronJobTimeOffset, err = time.ParseDuration(cronJobTimeOffsetDuration)
+	if err != nil {
+		log.Fatalf("Error parsing cron-job-time-offset-minute: %v", err)
+	}
+
 	log.SetFormatter(&log.TextFormatter{
 		DisableColors:    config.LogDisableColors,
 		DisableTimestamp: config.LogDisableTimestamp,

federator/influxdb/influxdb.go

@@ -118,7 +118,7 @@ func (b Backend) addBytesMeasurement(ctx context.Context, measurement string, tr
 		AddTag(remoteServiceHostgroupTag, trafficBandwidth.RemoteHostgroup).
 		AddTag(remoteServiceAddressTag, trafficBandwidth.RemoteDomain).
 		AddField(bandwidthBpsField, trafficBandwidth.BitsPerSecond).
-		SetTime(time.Now())
+		SetTime(t)
 	b.writeAPI.WritePoint(dataPoint)
 
 	return nil