Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Introduce ability to specify strategies for target allocation #1068

Closed
wants to merge 20 commits into from
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions apis/v1alpha1/opentelemetrycollector_types.go
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,8 @@ type OpenTelemetryCollectorSpec struct {

// OpenTelemetryTargetAllocator defines the configurations for the Prometheus target allocator.
type OpenTelemetryTargetAllocator struct {
// AllocationStrategy determines which strategy the target allocator should use for allocation
AllocationStrategy string `json:"allocationStrategy,omitempty"`
// ServiceAccount indicates the name of an existing service account to use with this instance.
// +optional
ServiceAccount string `json:"serviceAccount,omitempty"`
Expand Down
2 changes: 1 addition & 1 deletion apis/v1alpha1/zz_generated.deepcopy.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Original file line number Diff line number Diff line change
Expand Up @@ -683,6 +683,10 @@ spec:
description: TargetAllocator indicates a value which determines whether
to spawn a target allocation resource or not.
properties:
allocationStrategy:
description: AllocationStrategy determines which strategy the
target allocator should use for allocation
type: string
enabled:
description: Enabled indicates whether to use a target allocation
mechanism for Prometheus targets or not.
Expand Down
179 changes: 30 additions & 149 deletions cmd/otel-allocator/allocation/allocator.go
Original file line number Diff line number Diff line change
@@ -1,25 +1,21 @@
package allocation

import (
"fmt"
"net/url"
"sync"

"github.com/go-logr/logr"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
"github.com/prometheus/common/model"

"github.com/open-telemetry/opentelemetry-operator/cmd/otel-allocator/allocation/strategy"
)

var (
collectorsAllocatable = promauto.NewGauge(prometheus.GaugeOpts{
Name: "opentelemetry_allocator_collectors_allocatable",
Help: "Number of collectors the allocator is able to allocate to.",
})
targetsPerCollector = promauto.NewGaugeVec(prometheus.GaugeOpts{
Name: "opentelemetry_allocator_targets_per_collector",
Help: "The number of targets for each collector.",
}, []string{"collector_name"})

timeToAssign = promauto.NewHistogramVec(prometheus.HistogramOpts{
Name: "opentelemetry_allocator_time_to_allocate",
Help: "The time it takes to allocate",
Expand All @@ -33,156 +29,61 @@ var (
Keep a Map of what each collector currently holds and update it based on new scrape target updates
*/

type TargetItem struct {
JobName string
Link LinkJSON
TargetURL string
Label model.LabelSet
Collector *collector
}

func (t TargetItem) hash() string {
return t.JobName + t.TargetURL + t.Label.Fingerprint().String()
}

// Create a struct that holds collector - and jobs for that collector
// This struct will be parsed into endpoint with collector and jobs info

type collector struct {
Name string
NumTargets int
}

// Allocator makes decisions to distribute work among
// a number of OpenTelemetry collectors based on the number of targets.
// Users need to call SetTargets when they have new targets in their
// clusters and call SetCollectors when the collectors have changed.
type Allocator struct {
// m protects collectors and targetItems for concurrent use.
m sync.RWMutex
collectors map[string]*collector // all current collectors
targetItems map[string]*TargetItem
m sync.RWMutex
state strategy.State

log logr.Logger
log logr.Logger
strategy strategy.Allocator
}

// TargetItems returns a shallow copy of the targetItems map.
func (allocator *Allocator) TargetItems() map[string]*TargetItem {
func (allocator *Allocator) TargetItems() map[string]strategy.TargetItem {
allocator.m.RLock()
defer allocator.m.RUnlock()
targetItemsCopy := make(map[string]*TargetItem)
for k, v := range allocator.targetItems {
targetItemsCopy := make(map[string]strategy.TargetItem)
for k, v := range allocator.state.TargetItems() {
targetItemsCopy[k] = v
}
return targetItemsCopy
}

// Collectors returns a shallow copy of the collectors map.
func (allocator *Allocator) Collectors() map[string]*collector {
func (allocator *Allocator) Collectors() map[string]strategy.Collector {
allocator.m.RLock()
defer allocator.m.RUnlock()
collectorsCopy := make(map[string]*collector)
for k, v := range allocator.collectors {
collectorsCopy := make(map[string]strategy.Collector)
for k, v := range allocator.state.Collectors() {
collectorsCopy[k] = v
}
return collectorsCopy
}

// findNextCollector finds the next collector with fewer number of targets.
// This method is called from within SetTargets and SetCollectors, whose caller
// acquires the needed lock.
func (allocator *Allocator) findNextCollector() *collector {
var col *collector
for _, v := range allocator.collectors {
// If the initial collector is empty, set the initial collector to the first element of map
if col == nil {
col = v
} else {
if v.NumTargets < col.NumTargets {
col = v
}
}
}
return col
}

// addTargetToTargetItems assigns a target to the next available collector and adds it to the allocator's targetItems
// This method is called from within SetTargets and SetCollectors, whose caller acquires the needed lock.
// This is only called after the collectors are cleared or when a new target has been found in the tempTargetMap
func (allocator *Allocator) addTargetToTargetItems(target *TargetItem) {
chosenCollector := allocator.findNextCollector()
targetItem := TargetItem{
JobName: target.JobName,
Link: LinkJSON{fmt.Sprintf("/jobs/%s/targets", url.QueryEscape(target.JobName))},
TargetURL: target.TargetURL,
Label: target.Label,
Collector: chosenCollector,
}
allocator.targetItems[targetItem.hash()] = &targetItem
chosenCollector.NumTargets++
targetsPerCollector.WithLabelValues(chosenCollector.Name).Set(float64(chosenCollector.NumTargets))
}

// getCollectorChanges returns the new and removed collectors respectively.
// This method is called from within SetCollectors, which acquires the needed lock.
func (allocator *Allocator) getCollectorChanges(collectors []string) ([]string, []string) {
var newCollectors []string
var removedCollectors []string
// Used as a set to check for removed collectors
tempCollectorMap := map[string]bool{}
for _, s := range collectors {
if _, found := allocator.collectors[s]; !found {
newCollectors = append(newCollectors, s)
}
tempCollectorMap[s] = true
}
for k := range allocator.collectors {
if _, found := tempCollectorMap[k]; !found {
removedCollectors = append(removedCollectors, k)
}
}
return newCollectors, removedCollectors
}

// SetTargets accepts a list of targets that will be used to make
// load balancing decisions. This method should be called when there are
// new targets discovered or existing targets are shutdown.
func (allocator *Allocator) SetTargets(targets []TargetItem) {
func (allocator *Allocator) SetTargets(targets []strategy.TargetItem) {
timer := prometheus.NewTimer(timeToAssign.WithLabelValues("SetTargets"))
defer timer.ObserveDuration()

allocator.m.Lock()
defer allocator.m.Unlock()

// Make the temp map for access
tempTargetMap := make(map[string]TargetItem, len(targets))
tempTargetMap := make(map[string]strategy.TargetItem, len(targets))
for _, target := range targets {
tempTargetMap[target.hash()] = target
}

// Check for removals
for k, target := range allocator.targetItems {
// if the old target is no longer in the new list, remove it
if _, ok := tempTargetMap[k]; !ok {
allocator.collectors[target.Collector.Name].NumTargets--
delete(allocator.targetItems, k)
targetsPerCollector.WithLabelValues(target.Collector.Name).Set(float64(allocator.collectors[target.Collector.Name].NumTargets))
}
}

// Check for additions
for k, target := range tempTargetMap {
// Do nothing if the item is already there
if _, ok := allocator.targetItems[k]; ok {
continue
} else {
// Assign new set of collectors with the one different name
allocator.addTargetToTargetItems(&target)
}
tempTargetMap[target.Hash()] = target
}
newState := strategy.NewState(allocator.state.Collectors(), tempTargetMap)
allocator.state = allocator.strategy.Allocate(allocator.state, newState)
}

// SetCollectors sets the set of collectors with key=collectorName, value=Collector object.
// SetCollectors sets the set of collectors with key=collectorName, value=CollectorName object.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think we need this comment change - it makes it seem like key == value.

// This method is called when Collectors are added or removed.
func (allocator *Allocator) SetCollectors(collectors []string) {
log := allocator.log.WithValues("component", "opentelemetry-targetallocator")
Expand All @@ -197,41 +98,21 @@ func (allocator *Allocator) SetCollectors(collectors []string) {

allocator.m.Lock()
defer allocator.m.Unlock()
newCollectors, removedCollectors := allocator.getCollectorChanges(collectors)
if len(newCollectors) == 0 && len(removedCollectors) == 0 {
log.Info("No changes to the collectors found")
return
}

// Clear existing collectors
for _, k := range removedCollectors {
delete(allocator.collectors, k)
targetsPerCollector.WithLabelValues(k).Set(0)
}
// Insert the new collectors
for _, i := range newCollectors {
allocator.collectors[i] = &collector{Name: i, NumTargets: 0}
}

// find targets which need to be redistributed
var redistribute []*TargetItem
for _, item := range allocator.targetItems {
for _, s := range removedCollectors {
if item.Collector.Name == s {
redistribute = append(redistribute, item)
}
newCollectors := map[string]strategy.Collector{}
for _, s := range collectors {
newCollectors[s] = strategy.Collector{
Name: s,
NumTargets: 0,
}
}
// Re-Allocate the existing targets
for _, item := range redistribute {
allocator.addTargetToTargetItems(item)
}
newState := strategy.NewState(newCollectors, allocator.state.TargetItems())
allocator.state = allocator.strategy.Allocate(allocator.state, newState)
}

func NewAllocator(log logr.Logger) *Allocator {
func NewAllocator(log logr.Logger, allocatorStrategy strategy.Allocator) *Allocator {
return &Allocator{
log: log,
collectors: make(map[string]*collector),
targetItems: make(map[string]*TargetItem),
log: log,
state: strategy.NewState(make(map[string]strategy.Collector), make(map[string]strategy.TargetItem)),
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we don't need/use the maps here, shouldn't the NewState() function be able to handle nil values and make the maps itself?

strategy: allocatorStrategy,
}
}
Loading