gather

README

How to use this package

Make sure nats is running. Both publisher and subscriber are open

// NATS streaming server
m.natsServer = nats.NewServer(nats.Config{FilestoreDir: m.natsPath})
if err := m.natsServer.Open(); err != nil {
    m.logger.Error("Failed to start nats streaming server", zap.Error(err))
    return err
}

publisher := nats.NewAsyncPublisher("nats-publisher")
if err := publisher.Open(); err != nil {
    m.logger.Error("Failed to connect to streaming server", zap.Error(err))
    return err
}

subscriber := nats.NewQueueSubscriber("nats-subscriber")
if err := subscriber.Open(); err != nil {
    m.logger.Error("Failed to connect to streaming server", zap.Error(err))
    return err
}

Make sure the scraperTargetStorageService is accessible

scraperTargetSvc influxdb.ScraperTargetStoreService = m.boltClient

Setup recorder, Make sure subscriber subscribes use the correct recorder with the correct write service

recorder := gather.PlatformWriter{
    Timeout: time.Millisecond * 30,
    Writer: writer,
}
subscriber.Subscribe(MetricsSubject, "", &RecorderHandler{
    Logger:   logger,
    Recorder: recorder,
})

Start the scheduler

scraperScheduler, err := gather.NewScheduler(10, m.logger, scraperTargetSvc, publisher, subscriber, 0, 0)
if err != nil {
    m.logger.Error("Failed to create scraper subscriber", zap.Error(err))
    return err
}

Documentation

Index

• type Metrics
• type MetricsCollection
• type MetricsSlice
  • func (ms MetricsSlice) Points() (models.Points, error)
  • func (ms MetricsSlice) Reader() (io.Reader, error)
• type Scheduler
  • func NewScheduler(log *zap.Logger, scrapeQueueLength int, scrapesInProgress int, ...) (*Scheduler, error)
  • func (s *Scheduler) Close()
• type Scraper

Types

type Metrics

type Metrics struct {
	Name      string                 `json:"name"`
	Tags      map[string]string      `json:"tags"`
	Fields    map[string]interface{} `json:"fields"`
	Timestamp time.Time              `json:"timestamp"`
	Type      dto.MetricType         `json:"type"`
}

Metrics is the default influx based metrics.

type MetricsCollection

type MetricsCollection struct {
	OrgID        platform.ID  `json:"orgID"`
	BucketID     platform.ID  `json:"bucketID"`
	MetricsSlice MetricsSlice `json:"metrics"`
}

MetricsCollection is the struct including metrics and other requirements.

type MetricsSlice

type MetricsSlice []Metrics

MetricsSlice is a slice of Metrics

func (MetricsSlice) Points

func (ms MetricsSlice) Points() (models.Points, error)

Points convert the MetricsSlice to model.Points

func (MetricsSlice) Reader

func (ms MetricsSlice) Reader() (io.Reader, error)

Reader returns an io.Reader that enumerates the metrics. All metrics are allocated into the underlying buffer.

type Scheduler

type Scheduler struct {
	Targets influxdb.ScraperTargetStoreService
	// Interval is between each metrics gathering event.
	Interval time.Duration
	// contains filtered or unexported fields
}

Scheduler is struct to run scrape jobs.

func NewScheduler

func NewScheduler(
	log *zap.Logger,
	scrapeQueueLength int,
	scrapesInProgress int,
	targets influxdb.ScraperTargetStoreService,
	writer storage.PointsWriter,
	interval time.Duration,
) (*Scheduler, error)

NewScheduler creates a new Scheduler and subscriptions for scraper jobs.

func (*Scheduler) Close

func (s *Scheduler) Close()

type Scraper

type Scraper interface {
	Gather(ctx context.Context, target influxdb.ScraperTarget) (collected MetricsCollection, err error)
}

Scraper gathers metrics from a scraper target.