sharding

Documentation

Index

• Constants
• Variables
• func GetOrUpdateShardFromConfigMap(kubeClient kubernetes.Interface, settingsMgr *settings.SettingsManager, ...) (int, error)
• func InferShard() (int, error)
• type ClusterFilterFunction
  • func GetClusterFilter(db db.ArgoDB, distributionFunction DistributionFunction, replicas, shard int) ClusterFilterFunction
• type ClusterSharding
  • func (sharding *ClusterSharding) Add(c *v1alpha1.Cluster)
  • func (sharding *ClusterSharding) AddApp(a *v1alpha1.Application)
  • func (sharding *ClusterSharding) Delete(clusterServer string)
  • func (sharding *ClusterSharding) DeleteApp(a *v1alpha1.Application)
  • func (sharding *ClusterSharding) GetAppDistribution() map[string]int
  • func (sharding *ClusterSharding) GetDistribution() map[string]int
  • func (sharding *ClusterSharding) Init(clusters *v1alpha1.ClusterList, apps *v1alpha1.ApplicationList)
  • func (s *ClusterSharding) IsManagedCluster(c *v1alpha1.Cluster) bool
  • func (sharding *ClusterSharding) Update(oldCluster *v1alpha1.Cluster, newCluster *v1alpha1.Cluster)
  • func (sharding *ClusterSharding) UpdateApp(a *v1alpha1.Application)
• type ClusterShardingCache
  • func GetClusterSharding(kubeClient kubernetes.Interface, settingsMgr *settings.SettingsManager, ...) (ClusterShardingCache, error)
  • func NewClusterSharding(_ db.ArgoDB, shard, replicas int, shardingAlgorithm string) ClusterShardingCache
• type DistributionFunction
  • func ConsistentHashingWithBoundedLoadsDistributionFunction(clusters clusterAccessor, apps appAccessor, replicas int) DistributionFunction
  • func GetDistributionFunction(clusters clusterAccessor, apps appAccessor, shardingAlgorithm string, ...) DistributionFunction
  • func LegacyDistributionFunction(replicas int) DistributionFunction
  • func NoShardingDistributionFunction() DistributionFunction
  • func RoundRobinDistributionFunction(clusters clusterAccessor, replicas int) DistributionFunction

Constants

const ShardControllerMappingKey = "shardControllerMapping"

Variables

var (
	HeartbeatDuration = env.ParseNumFromEnv(common.EnvControllerHeartbeatTime, 10, 10, 60)
	HeartbeatTimeout  = 3 * HeartbeatDuration
)

Functions

func GetOrUpdateShardFromConfigMap

func GetOrUpdateShardFromConfigMap(kubeClient kubernetes.Interface, settingsMgr *settings.SettingsManager, replicas, shard int) (int, error)

GetOrUpdateShardFromConfigMap finds the shard number from the shard mapping configmap. If the shard mapping configmap does not exist, the function creates the shard mapping configmap. The function takes the shard number from the environment variable (default value -1, if not set) and passes it to this function. If the shard value passed to this function is -1, that is, the shard was not set as an environment variable, we default the shard number to 0 for computing the default config map.

func InferShard

func InferShard() (int, error)

InferShard extracts the shard index based on its hostname.

Types

type ClusterFilterFunction

type ClusterFilterFunction func(c *v1alpha1.Cluster) bool

func GetClusterFilter

func GetClusterFilter(db db.ArgoDB, distributionFunction DistributionFunction, replicas, shard int) ClusterFilterFunction

GetClusterFilter returns a ClusterFilterFunction which is a function taking a cluster as a parameter and returns whether or not the cluster should be processed by a given shard. It calls the distributionFunction to determine which shard will process the cluster, and if the given shard is equal to the calculated shard the function will return true.

type ClusterSharding

type ClusterSharding struct {
	Shard    int
	Replicas int
	Shards   map[string]int
	Clusters map[string]*v1alpha1.Cluster
	Apps     map[string]*v1alpha1.Application
	// contains filtered or unexported fields
}

func (*ClusterSharding) Add

func (sharding *ClusterSharding) Add(c *v1alpha1.Cluster)

func (*ClusterSharding) AddApp

func (sharding *ClusterSharding) AddApp(a *v1alpha1.Application)

func (*ClusterSharding) Delete

func (sharding *ClusterSharding) Delete(clusterServer string)

func (*ClusterSharding) DeleteApp

func (sharding *ClusterSharding) DeleteApp(a *v1alpha1.Application)

func (*ClusterSharding) GetAppDistribution

func (sharding *ClusterSharding) GetAppDistribution() map[string]int

GetAppDistribution should be not be called from a DestributionFunction because it could cause a deadlock when updateDistribution is called.

func (*ClusterSharding) GetDistribution

func (sharding *ClusterSharding) GetDistribution() map[string]int

func (*ClusterSharding) Init

func (sharding *ClusterSharding) Init(clusters *v1alpha1.ClusterList, apps *v1alpha1.ApplicationList)

func (*ClusterSharding) IsManagedCluster

func (s *ClusterSharding) IsManagedCluster(c *v1alpha1.Cluster) bool

IsManagedCluster returns whether or not the cluster should be processed by a given shard.

func (*ClusterSharding) Update

func (sharding *ClusterSharding) Update(oldCluster *v1alpha1.Cluster, newCluster *v1alpha1.Cluster)

func (*ClusterSharding) UpdateApp

func (sharding *ClusterSharding) UpdateApp(a *v1alpha1.Application)

type ClusterShardingCache

type ClusterShardingCache interface {
	Init(clusters *v1alpha1.ClusterList, apps *v1alpha1.ApplicationList)
	Add(c *v1alpha1.Cluster)
	Delete(clusterServer string)
	Update(oldCluster *v1alpha1.Cluster, newCluster *v1alpha1.Cluster)
	AddApp(a *v1alpha1.Application)
	DeleteApp(a *v1alpha1.Application)
	UpdateApp(a *v1alpha1.Application)
	IsManagedCluster(c *v1alpha1.Cluster) bool
	GetDistribution() map[string]int
	GetAppDistribution() map[string]int
}

func GetClusterSharding

func GetClusterSharding(kubeClient kubernetes.Interface, settingsMgr *settings.SettingsManager, shardingAlgorithm string, enableDynamicClusterDistribution bool) (ClusterShardingCache, error)

func NewClusterSharding

func NewClusterSharding(_ db.ArgoDB, shard, replicas int, shardingAlgorithm string) ClusterShardingCache

type DistributionFunction

type DistributionFunction func(c *v1alpha1.Cluster) int

func ConsistentHashingWithBoundedLoadsDistributionFunction

func ConsistentHashingWithBoundedLoadsDistributionFunction(clusters clusterAccessor, apps appAccessor, replicas int) DistributionFunction

ConsistentHashingWithBoundedLoadsDistributionFunction returns a DistributionFunction using an almost homogeneous distribution algorithm: for a given cluster the function will return the shard number based on a consistent hashing with bounded loads algorithm. This function ensures an almost homogenous distribution: each shards got assigned the fairly similar number of clusters +/-10% , but with it is resilient to sharding and/or number of clusters changes.

func GetDistributionFunction

func GetDistributionFunction(clusters clusterAccessor, apps appAccessor, shardingAlgorithm string, replicasCount int) DistributionFunction

GetDistributionFunction returns which DistributionFunction should be used based on the passed algorithm and the current datas.

func LegacyDistributionFunction

func LegacyDistributionFunction(replicas int) DistributionFunction

LegacyDistributionFunction returns a DistributionFunction using a stable distribution algorithm: for a given cluster the function will return the shard number based on the cluster id. This function is lightweight and can be distributed easily, however, it does not ensure an homogenous distribution as some shards may get assigned more clusters than others. It is the legacy function distribution that is kept for compatibility reasons

func NoShardingDistributionFunction

func NoShardingDistributionFunction() DistributionFunction

NoShardingDistributionFunction returns a DistributionFunction that will process all cluster by shard 0 the function is created for API compatibility purposes and is not supposed to be activated.

func RoundRobinDistributionFunction

func RoundRobinDistributionFunction(clusters clusterAccessor, replicas int) DistributionFunction