planprocessor

Documentation

Index

• type Config
  • func (cfg *Config) RegisterFlags(prefix string, f *flag.FlagSet)
• type PlanProcessor
• type Service
  • func NewService(cfg Config, plansDirectory string, bucket objstore.Bucket, ...) (*Service, error)

Types

type Config

type Config struct {
	// Exported config options.
	Name              string
	HeartbeatPeriod   time.Duration
	SchedulerEndpoint string
	NextPlanInterval  time.Duration
	GrpcConfig        grpcclient.Config
}

func (*Config) RegisterFlags

func (cfg *Config) RegisterFlags(prefix string, f *flag.FlagSet)

type PlanProcessor

type PlanProcessor interface {
	// Returns "id" that is appended to "finished" status file.
	ProcessPlanEntries(ctx context.Context, entries chan blocksconvert.PlanEntry) (string, error)
}

type Service

type Service struct {
	services.Service
	// contains filtered or unexported fields
}

This service implements common behaviour for plan-processing: 1) wait for next plan, 2) download plan, 3) process each plan entry, 4) delete local plan, 5) repeat. It gets plans from scheduler. During plan processing, this service maintains "progress" status file, and when plan processing finishes, it uploads "finished" plan.

func NewService

func NewService(cfg Config, plansDirectory string, bucket objstore.Bucket, cleanup func(logger log.Logger) error, factory func(planLog log.Logger, userID string, dayStart, dayEnd time.Time) PlanProcessor, l log.Logger, reg prometheus.Registerer) (*Service, error)

Creates new plan processor service. PlansDirectory is used for storing plan files. Bucket client used for downloading plan files. Cleanup function called on startup and after each build. Can be nil. Factory for creating PlanProcessor. Called for each new plan.