Merge pull request #4 from williamchanrico/add_upstream_downstream_services

Add upstream downstream services

Author: williamchanrico

cmd/planet-federator/internal/internal.go

@@ -72,7 +72,15 @@ func (s Service) Run(ctx context.Context) error {
 	cronScheduler := cron.New(cron.WithSeconds())
 	_, err := cronScheduler.AddFunc(s.Config.CronJobSchedule, s.TrafficBandwidthJobFunc)
 	if err != nil {
-		return fmt.Errorf("Error adding function to Cron scheduler: %v", err)
+		return fmt.Errorf("Error adding TrafficBandwidthJobFunc function to Cron scheduler: %v", err)
+	}
+	_, err = cronScheduler.AddFunc(s.Config.CronJobSchedule, s.UpstreamServicesJobFunc)
+	if err != nil {
+		return fmt.Errorf("Error adding UpstreamServicesJobFunc function to Cron scheduler: %v", err)
+	}
+	_, err = cronScheduler.AddFunc(s.Config.CronJobSchedule, s.DownstreamServicesJobFunc)
+	if err != nil {
+		return fmt.Errorf("Error adding DownstreamServicesJobFunc function to Cron scheduler: %v", err)
 	}
 	cronScheduler.Start()
 
@@ -137,3 +145,61 @@ func (s Service) TrafficBandwidthJobFunc() {
 
 	log.Infof("Traffic Bandwidth Job took: %v", time.Since(jobStartTime))
 }
+
+// UpstreamServicesJobFunc queries upstream services (planet-exporter) data from Prometheus and store
+// them in federator backend
+func (s Service) UpstreamServicesJobFunc() {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(s.Config.CronJobTimeoutSecond)*time.Second)
+	defer cancel()
+
+	jobStartTime := time.Now()
+	log.Debugf("A job started: %v", jobStartTime)
+
+	upstreamServices, err := s.PrometheusSvc.QueryPlanetExporterUpstreamServices(ctx, time.Now().Add(-15*time.Second), time.Now())
+	if err != nil {
+		log.Errorf("Error querying upstream services from prometheus: %v", err)
+	}
+
+	for _, svc := range upstreamServices {
+		_ = s.FederatorSvc.AddUpstreamService(ctx, federator.UpstreamService{
+			LocalHostgroup:    svc.LocalHostgroup,
+			LocalAddress:      svc.LocalAddress,
+			LocalProcessName:  svc.LocalProcessName,
+			UpstreamPort:      svc.UpstreamPort,
+			UpstreamHostgroup: svc.UpstreamHostgroup,
+			UpstreamAddress:   svc.UpstreamAddress,
+			Protocol:          svc.Protocol,
+		}, time.Now())
+	}
+
+	log.Infof("Upstream Service Job took: %v", time.Since(jobStartTime))
+}
+
+// DownstreamServicesJobFunc queries downstream services (planet-exporter) data from Prometheus and store
+// them in federator backend
+func (s Service) DownstreamServicesJobFunc() {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(s.Config.CronJobTimeoutSecond)*time.Second)
+	defer cancel()
+
+	jobStartTime := time.Now()
+	log.Debugf("A job started: %v", jobStartTime)
+
+	downstreamServices, err := s.PrometheusSvc.QueryPlanetExporterDownstreamServices(ctx, time.Now().Add(-15*time.Second), time.Now())
+	if err != nil {
+		log.Errorf("Error querying downstream services from prometheus: %v", err)
+	}
+
+	for _, svc := range downstreamServices {
+		_ = s.FederatorSvc.AddDownstreamService(ctx, federator.DownstreamService{
+			LocalHostgroup:      svc.LocalHostgroup,
+			LocalAddress:        svc.LocalAddress,
+			LocalProcessName:    svc.LocalProcessName,
+			LocalPort:           svc.LocalPort,
+			DownstreamHostgroup: svc.DownstreamHostgroup,
+			DownstreamAddress:   svc.DownstreamAddress,
+			Protocol:            svc.Protocol,
+		}, time.Now())
+	}
+
+	log.Infof("Downstream Service Job took: %v", time.Since(jobStartTime))
+}

federator/federator.go

@@ -22,7 +22,8 @@ import (
 // Federator package handles storing pre-processed planet-exporter data from Prometheus to
 // Influxdb and/or other time-series databases.
 
-// TrafficBandwidth represents services involved in an ingress/egress traffic
+// TrafficBandwidth represents a pair of services that are involved in an ingress/egress traffic
+// e.g. LocalHostgroup testapp transmit 10Mbps to RemoteHostgroup abc
 type TrafficBandwidth struct {
 	LocalHostgroup  string
 	LocalAddress    string
@@ -32,9 +33,40 @@ type TrafficBandwidth struct {
 	Direction       string
 }
 
-// Backend interface for a time-series DB handling pre-processed planet-exporter data
+// UpstreamService represents a target upstream service dependency of a local service process
+// e.g. LocalHostgroup testapp depends on UpstreamHostgroup abc, on abc's port 9000 via TCP protocol.
+//      LocalHostgroup -> UpstreamHostgroup:UpstreamPort
+//      testapp        -> abc:9000 (upstream)
+type UpstreamService struct {
+	LocalHostgroup    string
+	LocalAddress      string
+	LocalProcessName  string
+	UpstreamPort      string
+	UpstreamHostgroup string
+	UpstreamAddress   string
+	Protocol          string
+}
+
+// DownstreamService represents a target downstream service that depends on local service process
+// e.g. LocalHostgroup testapp has a dependency DownstreamHostgroup abc, on testapp's port 80 via TCP protocol.
+//      LocalHostgroup:LocalPort <- DownstreamHostgroup
+//      testapp:80               <- abc (downstream)
+type DownstreamService struct {
+	LocalHostgroup      string
+	LocalAddress        string
+	LocalProcessName    string
+	LocalPort           string
+	DownstreamHostgroup string
+	DownstreamAddress   string
+	Protocol            string
+}
+
+// Backend interface for a time-series DB that is handling pre-processed planet-exporter data
+// Planet Expoter <- Prometheus -> Planet Federator (pre-process) -> Time-series DB
 type Backend interface {
 	AddTrafficBandwidthData(context.Context, TrafficBandwidth, time.Time) error
+	AddUpstreamService(context.Context, UpstreamService, time.Time) error
+	AddDownstreamService(context.Context, DownstreamService, time.Time) error
 	Flush()
 }
 
@@ -55,6 +87,16 @@ func (s Service) AddTrafficBandwidthData(ctx context.Context, trafficBandwidth T
 	return s.backend.AddTrafficBandwidthData(ctx, trafficBandwidth, t)
 }
 
+// AddUpstreamService adds an upstream of a local service
+func (s Service) AddUpstreamService(ctx context.Context, upstreamService UpstreamService, t time.Time) error {
+	return s.backend.AddUpstreamService(ctx, upstreamService, t)
+}
+
+// AddDownstreamService adds a downstream of a local service
+func (s Service) AddDownstreamService(ctx context.Context, downstreamService DownstreamService, t time.Time) error {
+	return s.backend.AddDownstreamService(ctx, downstreamService, t)
+}
+
 // Flush any buffers related to backend
 func (s Service) Flush() {
 	s.backend.Flush()

federator/influxdb/influxdb.go

@@ -22,6 +22,7 @@ import (
 
 	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
 	influxdb2api "github.com/influxdata/influxdb-client-go/v2/api"
+	log "github.com/sirupsen/logrus"
 )
 
 // Backend interface for a time-series DB handling pre-processed planet-exporter data
@@ -34,26 +35,56 @@ type Backend struct {
 
 // New returns new influxdb federator backend
 func New(c influxdb2.Client, org, bucket string) Backend {
+	writeAPI := c.WriteAPI(org, bucket)
+
+	errChan := writeAPI.Errors()
+	go func() {
+		for err := range errChan {
+			log.Errorf("Async error received on influxdb writes API: %v", err)
+		}
+	}()
+
 	return Backend{
 		client:   c,
-		writeAPI: c.WriteAPI(org, bucket),
+		writeAPI: writeAPI,
 		org:      org,
 		bucket:   bucket,
 	}
 }
 
 const (
+	// Measurements
+
+	upstreamServiceMeasurement   = "upstream"
+	downstreamServiceMeasurement = "downstream"
+
 	ingressDirectionMeasurement = "ingress"
 	egressDirectionMeasurement  = "egress"
 	unknownDirectionMeasurement = "unknown"
 
-	localServiceHostgroupTag = "service"
-	localServiceAddressTag   = "address"
+	// Tags
+
+	localServiceHostgroupTag   = "service"
+	localServiceAddressTag     = "address"
+	localServicePortTag        = "port"
+	localServiceProcessNameTag = "process_name"
 
 	remoteServiceHostgroupTag = "remote_service"
 	remoteServiceAddressTag   = "remote_address"
 
-	bandwidthBpsField = "bandwidth_bps"
+	upstreamServiceHostgroupTag = "upstream_service"
+	upstreamServiceAddressTag   = "upstream_address"
+	upstreamServicePortTag      = "upstream_port"
+
+	downstreamServiceHostgroupTag = "upstream_service"
+	downstreamServiceAddressTag   = "upstream_address"
+
+	protocolTag = "protocol"
+
+	// Fields
+
+	bandwidthBpsField      = "bandwidth_bps"
+	serviceDependencyField = "service_dependency"
 )
 
 // AddTrafficBandwidthData adds a service's ingress bytes data point
@@ -93,6 +124,40 @@ func (b Backend) addBytesMeasurement(ctx context.Context, measurement string, tr
 	return nil
 }
 
+// AddUpstreamService adds an upstream service dependency of a service
+func (b Backend) AddUpstreamService(ctx context.Context, upstreamService federator.UpstreamService, t time.Time) error {
+	dataPoint := influxdb2.NewPointWithMeasurement(upstreamServiceMeasurement).
+		AddTag(localServiceHostgroupTag, upstreamService.LocalHostgroup).
+		AddTag(localServiceAddressTag, upstreamService.LocalAddress).
+		AddTag(upstreamServiceHostgroupTag, upstreamService.UpstreamHostgroup).
+		AddTag(upstreamServiceAddressTag, upstreamService.UpstreamAddress).
+		AddTag(upstreamServicePortTag, upstreamService.UpstreamPort).
+		AddTag(localServiceProcessNameTag, upstreamService.LocalProcessName).
+		AddTag(protocolTag, upstreamService.Protocol).
+		AddField(serviceDependencyField, 1).
+		SetTime(t)
+	b.writeAPI.WritePoint(dataPoint)
+
+	return nil
+}
+
+// AddDownstreamService adds a downstream service dependency of a service
+func (b Backend) AddDownstreamService(ctx context.Context, downstreamService federator.DownstreamService, t time.Time) error {
+	dataPoint := influxdb2.NewPointWithMeasurement(downstreamServiceMeasurement).
+		AddTag(localServiceHostgroupTag, downstreamService.LocalHostgroup).
+		AddTag(localServiceAddressTag, downstreamService.LocalAddress).
+		AddTag(localServicePortTag, downstreamService.LocalPort).
+		AddTag(localServiceProcessNameTag, downstreamService.LocalProcessName).
+		AddTag(downstreamServiceHostgroupTag, downstreamService.DownstreamHostgroup).
+		AddTag(downstreamServiceAddressTag, downstreamService.DownstreamAddress).
+		AddTag(protocolTag, downstreamService.Protocol).
+		AddField(serviceDependencyField, 1).
+		SetTime(t)
+	b.writeAPI.WritePoint(dataPoint)
+
+	return nil
+}
+
 // Flush all influxdb writes
 func (b Backend) Flush() {
 	b.writeAPI.Flush()