extractor

Documentation

Index

• Constants
• Variables
• type Metrics
  • func (s Metrics) Metrics(p *metrics.Prometheus) []prometheus.Collector
  • func (s Metrics) Refresh()
• type PlatformExtractor
  • func NewPlatformExtractor(batchSize uint32, continuousPulseRetrievingHalfPulseSeconds uint32, ...) *PlatformExtractor
  • func (e *PlatformExtractor) GetJetDrops(ctx context.Context) <-chan *types.PlatformPulseData
  • func (e *PlatformExtractor) LoadJetDrops(ctx context.Context, fromPulseNumber int64, toPulseNumber int64) error
  • func (e *PlatformExtractor) Start(ctx context.Context) error
  • func (e *PlatformExtractor) Stop(ctx context.Context) error
• type PlatformPulseExtractor
  • func NewPlatformPulseExtractor(client exporter.PulseExporterClient) *PlatformPulseExtractor
  • func (ppe *PlatformPulseExtractor) GetCurrentPulse(ctx context.Context) (uint32, error)
  • func (ppe *PlatformPulseExtractor) GetNextFinalizedPulse(ctx context.Context, p int64) (*exporter.FullPulse, error)

Constants

const LabelType = "type"

const PlatformAPIVersion = "2"

Variables

var (
	ErrorTypeNotFound          = prometheus.Labels{LabelType: "not_found"}
	ErrorTypeOnRecordExport    = prometheus.Labels{LabelType: "record_export"}
	ErrorTypeRateLimitExceeded = prometheus.Labels{LabelType: "rate_limit_exceeded"}

	ExtractProcessCount = prometheus.NewGauge(prometheus.GaugeOpts{
		Name: "gbe_extractor_process_count",
		Help: "The number of processes fetching data from heavy",
	})
	FromExtractorDataQueue = prometheus.NewGauge(prometheus.GaugeOpts{
		Name: "gbe_extractor_data_queue",
		Help: "The number of elements in extractor export data queue",
	})
	LastPulseFetched = prometheus.NewGauge(prometheus.GaugeOpts{
		Name: "gbe_extractor_last_pulse",
		Help: "The number of last pulse fetched by NextFinalizedPulse",
	})
	Errors = prometheus.NewCounterVec(prometheus.CounterOpts{
		Name: "gbe_extractor_errors",
		Help: "The number of errors received during data fetching",
	},
		[]string{LabelType},
	)
	ReceivedPulses = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "gbe_extractor_received_pulses",
		Help: "The number of pulses received",
	})
	ReceivedRecords = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "gbe_extractor_received_records",
		Help: "The number of records received",
	})

	RetrievePulsesCount = prometheus.NewGauge(prometheus.GaugeOpts{
		Name: "gbe_extractor_retrieve_pulses_count",
		Help: "The number of retrievePulses goroutines",
	})
	RetrieveRecordsCount = prometheus.NewGauge(prometheus.GaugeOpts{
		Name: "gbe_extractor_retrieve_records_count",
		Help: "The number of retrievePulses goroutines",
	})
)

Types

type Metrics

type Metrics struct{}

func (Metrics) Metrics

func (s Metrics) Metrics(p *metrics.Prometheus) []prometheus.Collector

func (Metrics) Refresh

func (s Metrics) Refresh()

type PlatformExtractor

type PlatformExtractor struct {
	// contains filtered or unexported fields
}

func NewPlatformExtractor

func NewPlatformExtractor(
	batchSize uint32,
	continuousPulseRetrievingHalfPulseSeconds uint32,
	maxWorkers int32,
	queueLen uint32,
	pulseExtractor interfaces.PulseExtractor,
	exporterClient exporter.RecordExporterClient,
	shutdownBE func(),
) *PlatformExtractor

func (*PlatformExtractor) GetJetDrops

func (e *PlatformExtractor) GetJetDrops(ctx context.Context) <-chan *types.PlatformPulseData

func (*PlatformExtractor) LoadJetDrops

func (e *PlatformExtractor) LoadJetDrops(ctx context.Context, fromPulseNumber int64, toPulseNumber int64) error

func (*PlatformExtractor) Start

func (e *PlatformExtractor) Start(ctx context.Context) error

func (*PlatformExtractor) Stop

func (e *PlatformExtractor) Stop(ctx context.Context) error

type PlatformPulseExtractor

type PlatformPulseExtractor struct {
	// contains filtered or unexported fields
}

func NewPlatformPulseExtractor

func NewPlatformPulseExtractor(client exporter.PulseExporterClient) *PlatformPulseExtractor

func (*PlatformPulseExtractor) GetCurrentPulse

func (ppe *PlatformPulseExtractor) GetCurrentPulse(ctx context.Context) (uint32, error)

func (*PlatformPulseExtractor) GetNextFinalizedPulse

func (ppe *PlatformPulseExtractor) GetNextFinalizedPulse(ctx context.Context, p int64) (*exporter.FullPulse, error)