ha

Documentation

Overview

Package ha implements a high availability clustering mode for the agent. It is also referred to as the "scraping service" mode, as this is a fairly accurate description of what it does: a series of configs are stored in a KV store and a cluster of agents pulls configs from the store and shards them amongst the cluster, thereby distributing scraping load.

Index

• Variables
• func GetCodec() codec.Codec
• type APIHandler
• type Config
  • func (c *Config) RegisterFlags(f *flag.FlagSet)
  • func (c *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet)
  • func (c *Config) UnmarshalYAML(unmarshal func(interface{}) error) error
• type ReadRing
• type Server
  • func New(reg prometheus.Registerer, cfg Config, globalConfig *config.GlobalConfig, ...) (*Server, error)
  • func (s *Server) AllConfigs(ctx context.Context) (<-chan instance.Config, error)
  • func (s *Server) DeleteConfiguration(r *http.Request) (interface{}, error)
  • func (s *Server) Flush()
  • func (s *Server) GetConfiguration(r *http.Request) (interface{}, error)
  • func (s *Server) ListConfigurations(r *http.Request) (interface{}, error)
  • func (s *Server) PutConfiguration(r *http.Request) (interface{}, error)
  • func (s *Server) Reshard(ctx context.Context, _ *agentproto.ReshardRequest) (_ *empty.Empty, err error)
  • func (s *Server) Stop() error
  • func (s *Server) TransferOut(ctx context.Context) error
  • func (s *Server) WireAPI(r *mux.Router)
  • func (s *Server) WireGRPC(srv *grpc.Server)
• type ShardingInstanceManager
  • func NewShardingInstanceManager(logger log.Logger, wrap instance.Manager, ring ReadRing, addr string) ShardingInstanceManager
  • func (m ShardingInstanceManager) ApplyConfig(c instance.Config) error
  • func (m ShardingInstanceManager) DeleteConfig(name string) error
  • func (m ShardingInstanceManager) ListConfigs() map[string]instance.Config
  • func (m ShardingInstanceManager) ListInstances() map[string]instance.ManagedInstance
  • func (m ShardingInstanceManager) Stop()

Variables

var (

	// DefaultConfig provides default values for the config
	DefaultConfig = *flagutil.DefaultConfigFromFlags(&Config{}).(*Config)
)

Functions

func GetCodec

func GetCodec() codec.Codec

GetCodec returns the codec for encoding and decoding instance.Configs

Types

type APIHandler

type APIHandler func(r *http.Request) (interface{}, error)

APIHandler is a function that returns a configapi Response type and optionally an error.

type Config

type Config struct {
	Enabled         bool                  `yaml:"enabled"`
	ReshardInterval time.Duration         `yaml:"reshard_interval"`
	ReshardTimeout  time.Duration         `yaml:"reshard_timeout"`
	KVStore         kv.Config             `yaml:"kvstore"`
	Lifecycler      ring.LifecyclerConfig `yaml:"lifecycler"`
}

Config describes how to instantiate a scraping service Server instance.

func (*Config) RegisterFlags

func (c *Config) RegisterFlags(f *flag.FlagSet)

RegisterFlags adds the flags required to config the Server to the given FlagSet.

func (*Config) RegisterFlagsWithPrefix

func (c *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet)

RegisterFlagsWithPrefix adds the flags required to config this to the given FlagSet with a specified prefix.

func (*Config) UnmarshalYAML

func (c *Config) UnmarshalYAML(unmarshal func(interface{}) error) error

UnmarshalYAML implements yaml.Unmarshaler.

type ReadRing

type ReadRing interface {
	http.Handler

	Get(key uint32, op ring.Operation, bufDescs []ring.InstanceDesc, bufHosts, bufZones []string) (ring.ReplicationSet, error)
	GetAllHealthy(op ring.Operation) (ring.ReplicationSet, error)
}

ReadRing is a subset of the Cortex ring.ReadRing interface with only the functionality used by the HA server.

type Server

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

Server implements the HA scraping service.

func New

func New(reg prometheus.Registerer, cfg Config, globalConfig *config.GlobalConfig, clientConfig client.Config, logger log.Logger, im instance.Manager) (*Server, error)

New creates a new HA scraping service instance.

func (*Server) AllConfigs

func (s *Server) AllConfigs(ctx context.Context) (<-chan instance.Config, error)

AllConfigs gets all configs known to the KV store.

func (*Server) DeleteConfiguration

func (s *Server) DeleteConfiguration(r *http.Request) (interface{}, error)

DeleteConfiguration deletes an existing named configuration.

func (*Server) Flush

func (s *Server) Flush()

Flush satisfies ring.FlushTransferer. It is a no-op for the Agent.

func (*Server) GetConfiguration

func (s *Server) GetConfiguration(r *http.Request) (interface{}, error)

GetConfiguration returns an existing named configuration.

func (*Server) ListConfigurations

func (s *Server) ListConfigurations(r *http.Request) (interface{}, error)

ListConfigurations returns a list of the named configurations or all configurations associated with the Prometheus agent.

func (*Server) PutConfiguration

func (s *Server) PutConfiguration(r *http.Request) (interface{}, error)

PutConfiguration creates or updates a named configuration. Completely overrides the previous configuration if it exists.

func (*Server) Reshard

func (s *Server) Reshard(ctx context.Context, _ *agentproto.ReshardRequest) (_ *empty.Empty, err error)

Reshard initiates an entire reshard of the current HA scraping service instance. All configs will be reloaded from the KV store and the scraping service instance will see what should be managed locally.

Satisfies agentproto.ScrapingServiceServer.

func (*Server) Stop

func (s *Server) Stop() error

Stop stops the HA server and its dependencies.

func (*Server) TransferOut

func (s *Server) TransferOut(ctx context.Context) error

TransferOut satisfies ring.FlushTransferer. It connects to all other healthy agents in the cluster and tells them to reshard.

func (*Server) WireAPI

func (s *Server) WireAPI(r *mux.Router)

WireAPI injects routes into the provided mux router for the config management API.

func (*Server) WireGRPC

func (s *Server) WireGRPC(srv *grpc.Server)

WireGRPC injects gRPC server handlers into the provided gRPC server.

type ShardingInstanceManager

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

ShardingInstanceManager wraps around an existing instance.Manager and uses a hash ring to determine if a config should be applied. If an applied config used to be owned by the local address but no longer does, it will be deleted on the next apply.

func NewShardingInstanceManager

func NewShardingInstanceManager(logger log.Logger, wrap instance.Manager, ring ReadRing, addr string) ShardingInstanceManager

NewShardingInstanceManager creates a new ShardingInstanceManager that wraps around an underlying instance.Manager. ring and addr are used together to do hash ring lookups; for a given applied config, it is owned by the instance of ShardingInstanceManager if looking up its hash in the ring results in the address specified by addr.

func (ShardingInstanceManager) ApplyConfig

func (m ShardingInstanceManager) ApplyConfig(c instance.Config) error

ApplyConfig implements instance.Manager.ApplyConfig.

func (ShardingInstanceManager) DeleteConfig

func (m ShardingInstanceManager) DeleteConfig(name string) error

DeleteConfig implements instance.Manager.DeleteConfig.

func (ShardingInstanceManager) ListConfigs

func (m ShardingInstanceManager) ListConfigs() map[string]instance.Config

ListConfigs returns the list of configs that have been applied through the ShardingInstanceManager. It will return a subset of the overall set of configs passed to the instance.Manager as a whole.

Returning the subset of configs that only the ShardingInstanceManager applied itself allows for the underlying instance.Manager to manage its own set of configs that will not be affected by the scraping service resharding and deleting configs that aren't found in the KV store.

func (ShardingInstanceManager) ListInstances

func (m ShardingInstanceManager) ListInstances() map[string]instance.ManagedInstance

ListInstances returns the list of instances that have been applied through the ShardingInstanceManager. It will return a subset of the overall set of configs passed to the instance.Manager as a whole.

Returning the subset of configs that only the ShardingInstanceManager applied itself allows for the underlying instance.Manager to manage its own set of configs that will not be affected by the scraping service resharding and deleting configs that aren't found in the KV store.

func (ShardingInstanceManager) Stop

func (m ShardingInstanceManager) Stop()

Stop implements instance.Manager.Stop. It only stops the configs passed through to this manager and not all instances.