Adds possibility to call BBS to retrieve stats for application's

processes states.

collectors/applications.go

@@ -224,6 +224,7 @@ func (c ApplicationsCollector) reportApp(application models.Application, objs *m
 			process = cProc
 		}
 	}
+
 	spaceRel, ok := application.Relationships[constant.RelationshipTypeSpace]
 	if !ok {
 		return fmt.Errorf("could not find space relation in application '%s'", application.GUID)
@@ -290,6 +291,19 @@ func (c ApplicationsCollector) reportApp(application models.Application, objs *m
 		string(application.State),
 	).Set(float64(process.Instances.Value))
 
+	runningInstances := appSum.RunningInstances
+	// Use bbs data if available
+	if len(objs.ProcessActualLRPs) > 0 {
+		runningsInstances := 0
+		lrps, ok := objs.ProcessActualLRPs[process.GUID]
+		if ok {
+			for _, lrp := range lrps {
+				if lrp.State == "RUNNING" {
+					runningsInstances++
+				}
+			}
+		}
+	}
 	c.applicationInstancesRunningMetric.WithLabelValues(
 		application.GUID,
 		application.Name,
@@ -298,7 +312,7 @@ func (c ApplicationsCollector) reportApp(application models.Application, objs *m
 		space.GUID,
 		space.Name,
 		string(application.State),
-	).Set(float64(appSum.RunningInstances))
+	).Set(float64(runningInstances))
 
 	c.applicationMemoryMbMetric.WithLabelValues(
 		application.GUID,

collectors/collectors.go

@@ -14,7 +14,8 @@ type ObjectCollector interface {
 
 type Collector struct {
 	workers    int
-	config     *fetcher.CFConfig
+	cfConfig   *fetcher.CFConfig
+	bbsConfig  *fetcher.BBSConfig
 	filter     *filters.Filter
 	collectors []ObjectCollector
 }
@@ -24,12 +25,14 @@ func NewCollector(
 	environment string,
 	deployment string,
 	workers int,
-	config *fetcher.CFConfig,
+	cfConfig *fetcher.CFConfig,
+	bbsConfig *fetcher.BBSConfig,
 	filter *filters.Filter,
 ) (*Collector, error) {
 	res := &Collector{
 		workers:    workers,
-		config:     config,
+		cfConfig:   cfConfig,
+		bbsConfig:  bbsConfig,
 		filter:     filter,
 		collectors: []ObjectCollector{},
 	}
@@ -118,8 +121,9 @@ func NewCollector(
 }
 
 func (c *Collector) Collect(ch chan<- prometheus.Metric) {
-	fetcher := fetcher.NewFetcher(c.workers, c.config, c.filter)
+	fetcher := fetcher.NewFetcher(c.workers, c.cfConfig, c.bbsConfig, c.filter)
 	objs := fetcher.GetObjects()
+
 	for _, collector := range c.collectors {
 		collector.Collect(objs, ch)
 	}

fetcher/bbs_client.go

@@ -0,0 +1,62 @@
+package fetcher
+
+import (
+	"strings"
+	"time"
+
+	"code.cloudfoundry.org/bbs"
+	"code.cloudfoundry.org/bbs/models"
+	"code.cloudfoundry.org/bbs/trace"
+	"code.cloudfoundry.org/lager/v3"
+)
+
+const (
+	clientSessionCacheSize int = -1
+	maxIdleConnsPerHost    int = -1
+)
+
+type BBSClient struct {
+	client bbs.Client
+	config *BBSConfig
+	logger lager.Logger
+}
+
+type BBSConfig struct {
+	URL            string `yaml:"url"`
+	Timeout        int    `yaml:"timeout"`
+	CAFile         string `yaml:"ca_file"`
+	CertFile       string `yaml:"cert_file"`
+	KeyFile        string `yaml:"key_file"`
+	SkipCertVerify bool   `yaml:"skip_cert_verify"`
+}
+
+func NewBBSClient(config *BBSConfig) (*BBSClient, error) {
+	var err error
+	bbsClient := BBSClient{
+		config: config,
+		logger: lager.NewLogger("bbs-client"),
+	}
+	bbsClientConfig := bbs.ClientConfig{
+		URL:            config.URL,
+		Retries:        1,
+		RequestTimeout: time.Duration(config.Timeout) * time.Second,
+	}
+	if strings.HasPrefix(config.URL, "https://") {
+		bbsClientConfig.IsTLS = true
+		bbsClientConfig.InsecureSkipVerify = config.SkipCertVerify
+		bbsClientConfig.CAFile = config.CAFile
+		bbsClientConfig.CertFile = config.CertFile
+		bbsClientConfig.KeyFile = config.KeyFile
+		bbsClientConfig.ClientSessionCacheSize = clientSessionCacheSize
+		bbsClientConfig.MaxIdleConnsPerHost = maxIdleConnsPerHost
+	}
+	bbsClient.client, err = bbs.NewClientWithConfig(bbsClientConfig)
+	return &bbsClient, err
+}
+
+func (b *BBSClient) GetActualLRPs() ([]*models.ActualLRP, error) {
+	traceID := trace.GenerateTraceID()
+	actualLRPs, err := b.client.ActualLRPs(b.logger, traceID, models.ActualLRPFilter{})
+
+	return actualLRPs, err
+}

fetcher/fetcher.go

@@ -36,14 +36,16 @@ type CFConfig struct {
 
 type Fetcher struct {
 	sync.Mutex
-	config *CFConfig
-	worker *Worker
+	cfConfig  *CFConfig
+	bbsConfig *BBSConfig
+	worker    *Worker
 }
 
-func NewFetcher(threads int, config *CFConfig, filter *filters.Filter) *Fetcher {
+func NewFetcher(threads int, config *CFConfig, bbsConfig *BBSConfig, filter *filters.Filter) *Fetcher {
 	return &Fetcher{
-		config: config,
-		worker: NewWorker(threads, filter),
+		cfConfig:  config,
+		bbsConfig: bbsConfig,
+		worker:    NewWorker(threads, filter),
 	}
 }
 
@@ -82,20 +84,27 @@ func (c *Fetcher) workInit() {
 	c.worker.PushIf("service_route_bindings", c.fetchServiceRouteBindings, filters.ServiceRouteBindings)
 	c.worker.PushIf("users", c.fetchUsers, filters.Events)
 	c.worker.PushIf("events", c.fetchEvents, filters.Events)
+	c.worker.PushIf("actual_lrps", c.fetchActualLRPs)
 }
 
 func (c *Fetcher) fetch() *models.CFObjects {
 	result := models.NewCFObjects()
 
-	session, err := NewSessionExt(c.config)
+	session, err := NewSessionExt(c.cfConfig)
 	if err != nil {
 		log.WithError(err).Error("unable to initialize cloud foundry clients")
 		result.Error = err
 		return result
 	}
+	bbs, err := NewBBSClient(c.bbsConfig)
+	if err != nil {
+		log.WithError(err).Error("unable to initialize bbs client")
+		result.Error = err
+		return result
+	}
 
 	c.workInit()
 
-	result.Error = c.worker.Do(session, result)
+	result.Error = c.worker.Do(session, bbs, result)
 	return result
 }