service

Documentation

Index

• Variables
• type CantabularClient
• type DatasetAPIClient
• type HTTPServer
• type HealthChecker
• type ImportAPIClient
• type RecipeAPIClient
• type Service
  • func New() *Service
  • func (svc *Service) Close(ctx context.Context) error
  • func (svc *Service) Init(ctx context.Context, cfg *config.Config, buildTime, gitCommit, version string) error
  • func (svc *Service) Start(ctx context.Context, svcErrors chan error) error

Variables

var GetCantabularClient = func(cfg *config.Config) CantabularClient {
	return cantabular.NewClient(
		cantabular.Config{
			Host:       cfg.CantabularURL,
			ExtApiHost: cfg.CantabularExtURL,
		},
		dphttp.NewClient(),
		nil,
	)
}

var GetDatasetAPIClient = func(cfg *config.Config) DatasetAPIClient {
	return dataset.NewAPIClient(cfg.DatasetAPIURL)
}

var GetHTTPServer = func(bindAddr string, router http.Handler) HTTPServer {
	s := dphttp.NewServer(bindAddr, router)
	s.HandleOSSignals = false
	return s
}

GetHTTPServer returns an http server

var GetHealthCheck = func(cfg *config.Config, buildTime, gitCommit, version string) (HealthChecker, error) {
	versionInfo, err := healthcheck.NewVersionInfo(buildTime, gitCommit, version)
	if err != nil {
		return nil, err
	}
	hc := healthcheck.New(versionInfo, cfg.HealthCheckCriticalTimeout, cfg.HealthCheckInterval)
	return &hc, nil
}

GetHealthCheck returns a healthcheck

var GetImportAPIClient = func(cfg *config.Config) ImportAPIClient {
	return importapi.New(cfg.ImportAPIURL)
}

var GetKafkaConsumer = func(ctx context.Context, cfg *config.Config) (kafka.IConsumerGroup, error) {
	kafkaOffset := kafka.OffsetNewest
	if cfg.KafkaConfig.OffsetOldest {
		kafkaOffset = kafka.OffsetOldest
	}
	cgConfig := &kafka.ConsumerGroupConfig{
		BrokerAddrs:       cfg.KafkaConfig.Addr,
		Topic:             cfg.KafkaConfig.InstanceStartedTopic,
		GroupName:         cfg.KafkaConfig.InstanceStartedGroup,
		MinBrokersHealthy: &cfg.KafkaConfig.ConsumerMinBrokersHealthy,
		KafkaVersion:      &cfg.KafkaConfig.Version,
		NumWorkers:        &cfg.KafkaConfig.NumWorkers,
		Offset:            &kafkaOffset,
	}
	if cfg.KafkaConfig.SecProtocol == config.KafkaTLSProtocolFlag {
		cgConfig.SecurityConfig = kafka.GetSecurityConfig(
			cfg.KafkaConfig.SecCACerts,
			cfg.KafkaConfig.SecClientCert,
			cfg.KafkaConfig.SecClientKey,
			cfg.KafkaConfig.SecSkipVerify,
		)
	}
	return kafka.NewConsumerGroup(ctx, cgConfig)
}

GetKafkaConsumer returns a Kafka consumer with the provided config

var GetKafkaProducer = func(ctx context.Context, cfg *config.Config) (kafka.IProducer, error) {
	pConfig := &kafka.ProducerConfig{
		BrokerAddrs:       cfg.KafkaConfig.Addr,
		Topic:             cfg.KafkaConfig.CategoryDimensionImportTopic,
		MinBrokersHealthy: &cfg.KafkaConfig.ProducerMinBrokersHealthy,
		KafkaVersion:      &cfg.KafkaConfig.Version,
		MaxMessageBytes:   &cfg.KafkaConfig.MaxBytes,
	}
	if cfg.KafkaConfig.SecProtocol == config.KafkaTLSProtocolFlag {
		pConfig.SecurityConfig = kafka.GetSecurityConfig(
			cfg.KafkaConfig.SecCACerts,
			cfg.KafkaConfig.SecClientCert,
			cfg.KafkaConfig.SecClientKey,
			cfg.KafkaConfig.SecSkipVerify,
		)
	}
	return kafka.NewProducer(ctx, pConfig)
}

GetKafkaProducer returns a kafka producer with the provided config

var GetRecipeAPIClient = func(cfg *config.Config) RecipeAPIClient {
	return recipe.NewClient(cfg.RecipeAPIURL)
}

Types

type CantabularClient

type CantabularClient interface {
	GetDimensionsByName(ctx context.Context, req cantabular.GetDimensionsByNameRequest) (*cantabular.GetDimensionsResponse, error)
	Checker(context.Context, *healthcheck.CheckState) error
	CheckerAPIExt(ctx context.Context, state *healthcheck.CheckState) error
}

type DatasetAPIClient

type DatasetAPIClient interface {
	PutInstance(context.Context, string, string, string, string, dataset.UpdateInstance, string) (string, error)
	PutInstanceState(context.Context, string, string, dataset.State, string) (string, error)
	Checker(context.Context, *healthcheck.CheckState) error
}

type HTTPServer

type HTTPServer interface {
	ListenAndServe() error
	Shutdown(ctx context.Context) error
}

HTTPServer defines the required methods from the HTTP server

type HealthChecker

type HealthChecker interface {
	Handler(w http.ResponseWriter, req *http.Request)
	Start(ctx context.Context)
	Stop()
	AddAndGetCheck(name string, checker healthcheck.Checker) (check *healthcheck.Check, err error)
	Subscribe(s healthcheck.Subscriber, checks ...*healthcheck.Check)
}

HealthChecker defines the required methods from Healthcheck

type ImportAPIClient

type ImportAPIClient interface {
	UpdateImportJobState(context.Context, string, string, importapi.State) error
}

type RecipeAPIClient

type RecipeAPIClient interface {
	GetRecipe(context.Context, string, string, string) (*recipe.Recipe, error)
	Checker(context.Context, *healthcheck.CheckState) error
}

type Service

type Service struct {
	Cfg         *config.Config
	Server      HTTPServer
	HealthCheck HealthChecker
	Consumer    kafka.IConsumerGroup
	Producer    kafka.IProducer
	// contains filtered or unexported fields
}

Service contains all the configs, server and clients to run the event handler service

func New

func New() *Service

New creates a new empty service

func (*Service) Close

func (svc *Service) Close(ctx context.Context) error

Close gracefully shuts the service down in the required order, with timeout

func (*Service) Init

func (svc *Service) Init(ctx context.Context, cfg *config.Config, buildTime, gitCommit, version string) error

Init initialises all the service dependencies, including healthcheck with checkers, api and middleware

func (*Service) Start

func (svc *Service) Start(ctx context.Context, svcErrors chan error) error

Start starts an initialised service