controller

Documentation

Index

• Variables
• type AlgChoose
  • func (alg AlgChoose) IsValid() bool
  • func (alg AlgChoose) String() string
• type Cluster
• type ClusterConfig
• type ClusterController
  • func NewClusterController(cfg *ClusterConfig, proxy proxy.Cacher, stopCh <-chan struct{}) (ClusterController, error)
• type HostIDC
• type ServiceConfig
• type ServiceController
  • func NewServiceController(cfg ServiceConfig, cmCli clustermgr.APIAccess, proxy proxy.Cacher, ...) (ServiceController, error)
• type Unit
• type VolumeGetter
  • func NewVolumeGetter(clusterID proto.ClusterID, service ServiceController, proxy proxy.Cacher, ...) (VolumeGetter, error)
• type VolumePhy

Variables

var (
	ErrNoSuchCluster      = errors.New("controller: no such cluster")
	ErrNoClusterAvailable = errors.New("controller: no cluster available")
	ErrInvalidChooseAlg   = errors.New("controller: invalid cluster chosen algorithm")
)

errors

Types

type AlgChoose

type AlgChoose uint32

AlgChoose algorithm of choose cluster

const (

	// AlgAvailable available capacity and some random alloc
	AlgAvailable AlgChoose
	// AlgRoundRobin alloc cluster round robin
	AlgRoundRobin
	// AlgRandom completely random alloc
	AlgRandom
)

func (AlgChoose) IsValid

func (alg AlgChoose) IsValid() bool

IsValid returns valid algorithm or not.

func (AlgChoose) String

func (alg AlgChoose) String() string

type Cluster

type Cluster struct {
	ClusterID proto.ClusterID `json:"cluster_id"`
	Hosts     []string        `json:"hosts"`
}

Cluster cluster config, each clusterID related to hosts list

type ClusterConfig

type ClusterConfig struct {
	IDC               string       `json:"-"` // passing by stream config.
	Region            string       `json:"region"`
	RegionMagic       string       `json:"region_magic"`
	ClusterReloadSecs int          `json:"cluster_reload_secs"`
	ServiceReloadSecs int          `json:"service_reload_secs"`
	CMClientConfig    cmapi.Config `json:"clustermgr_client_config"`

	ServicePunishThreshold      uint32 `json:"service_punish_threshold"`
	ServicePunishValidIntervalS int    `json:"service_punish_valid_interval_s"`

	ConsulAgentAddr string    `json:"consul_agent_addr"`
	Clusters        []Cluster `json:"clusters"`
}

ClusterConfig cluster config

Region and RegionMagic are paired, magic cannot change if one region was deployed.

type ClusterController

type ClusterController interface {
	// Region returns region in configuration
	Region() string
	// All returns all cluster info in this region
	All() []*cmapi.ClusterInfo
	// ChooseOne returns a available cluster to upload
	ChooseOne() (*cmapi.ClusterInfo, error)
	// GetServiceController return ServiceController in specified cluster
	GetServiceController(clusterID proto.ClusterID) (ServiceController, error)
	// GetVolumeGetter return VolumeGetter in specified cluster
	GetVolumeGetter(clusterID proto.ClusterID) (VolumeGetter, error)
	// GetConfig get specified config of key from cluster manager
	GetConfig(ctx context.Context, key string) (string, error)
	// ChangeChooseAlg change alloc algorithm
	ChangeChooseAlg(alg AlgChoose) error
}

ClusterController controller of clusters in one region

func NewClusterController

func NewClusterController(cfg *ClusterConfig, proxy proxy.Cacher, stopCh <-chan struct{}) (ClusterController, error)

NewClusterController returns a cluster controller

type HostIDC

type HostIDC struct {
	Host     string
	IDC      string
	Punished bool
}

HostIDC item of host with idc

type ServiceConfig

type ServiceConfig struct {
	ClusterID                   proto.ClusterID
	IDC                         string
	ReloadSec                   int
	LoadDiskInterval            int
	ServicePunishThreshold      uint32
	ServicePunishValidIntervalS int
}

ServiceConfig service config

type ServiceController

type ServiceController interface {
	// GetServiceHost return an available service host
	GetServiceHost(ctx context.Context, name string) (host string, err error)
	// GetServiceHosts return all available service random sorted hosts
	GetServiceHosts(ctx context.Context, name string) (hosts []string, err error)
	// GetDiskHost return an disk's related data node host
	GetDiskHost(ctx context.Context, diskID proto.DiskID) (hostIDC *HostIDC, err error)
	// PunishService will punish an service host for an punishTimeSec interval
	PunishService(ctx context.Context, service, host string, punishTimeSec int)
	// PunishServiceWithThreshold will punish an service host for
	// an punishTimeSec interval if service failed times satisfied with threshold during some interval time
	PunishServiceWithThreshold(ctx context.Context, service, host string, punishTimeSec int)
	// PunishDisk will punish a disk host for an punishTimeSec interval
	PunishDisk(ctx context.Context, diskID proto.DiskID, punishTimeSec int)
	// PunishDiskWithThreshold will punish a disk host for
	// an punishTimeSec interval if disk host failed times satisfied with threshold
	PunishDiskWithThreshold(ctx context.Context, diskID proto.DiskID, punishTimeSec int)
}

ServiceController support for both data node discovery and normal service discovery

func NewServiceController

func NewServiceController(cfg ServiceConfig, cmCli clustermgr.APIAccess, proxy proxy.Cacher,
	stopCh <-chan struct{}) (ServiceController, error)

NewServiceController returns a service controller

type Unit

type Unit = clustermgr.Unit

Unit alias of clustermgr.Unit

type VolumeGetter

type VolumeGetter interface {
	// Get returns volume physical location of vid
	Get(ctx context.Context, vid proto.Vid, isCache bool) *VolumePhy
	// Punish punish vid with interval seconds
	Punish(ctx context.Context, vid proto.Vid, punishIntervalS int)
}

VolumeGetter getter of volume physical location

ctx: context with trace or something

isCache: is false means reading from proxy cluster then updating memcache

otherwise reading from memcache -> proxy -> cluster

func NewVolumeGetter

func NewVolumeGetter(clusterID proto.ClusterID, service ServiceController,
	proxy proxy.Cacher, memExpiration time.Duration) (VolumeGetter, error)

NewVolumeGetter new a volume getter

memExpiration expiration of memcache, 0 means no expiration

type VolumePhy

type VolumePhy struct {
	Vid       proto.Vid
	CodeMode  codemode.CodeMode
	IsPunish  bool
	Version   uint32
	Timestamp int64
	Units     []Unit
}

VolumePhy volume physical info

Vid, CodeMode and Units are from cluster
IsPunish is cached in memory
Version is versioned in proxy
Timestamp is cached in proxy to clear outdate volume