func()

in pkg/scheduler/placement/tag_rule.go [78:150]

• 60 lines of code
• 16 McCabe index (conditional complexity)

func (tr *tagRule) placeApplication(app *objects.Application, queueFn func(string) *objects.Queue) (string, error) {
	// if the tag is not present we can skipp all other processing
	tagVal := app.GetTag(tr.tagName)
	if tagVal == "" {
		return "", nil
	}
	// before anything run the filter
	if !tr.filter.allowUser(app.GetUser()) {
		log.Log(log.SchedApplication).Debug("Tag rule filtered",
			zap.String("application", app.ApplicationID),
			zap.Any("user", app.GetUser()),
			zap.String("tagName", tr.tagName))
		return "", nil
	}
	var parentName string
	var err error
	queueName := tagVal
	// fully qualified queue, do not run the parent rule
	if strings.HasPrefix(queueName, configs.RootQueue+configs.DOT) {
		parts := strings.Split(queueName, configs.DOT)
		for _, part := range parts {
			if err = configs.IsQueueNameValid(part); err != nil {
				return "", err
			}
		}
	} else {
		// not fully qualified queue
		childQueueName := replaceDot(tagVal)
		if err = configs.IsQueueNameValid(childQueueName); err != nil {
			return "", err
		}
		// run the parent rule if set
		if tr.parent != nil {
			parentName, err = tr.parent.placeApplication(app, queueFn)
			// failed parent rule, fail this rule
			if err != nil {
				return "", err
			}
			// rule did not match: this could be filter or create flag related
			if parentName == "" {
				return "", nil
			}
			// check if this is a parent queue and qualify it
			if !strings.HasPrefix(parentName, configs.RootQueue+configs.DOT) {
				parentName = configs.RootQueue + configs.DOT + parentName
			}
			// if the parent queue exists it cannot be a leaf
			parentQueue := queueFn(parentName)
			if parentQueue != nil && parentQueue.IsLeafQueue() {
				return "", fmt.Errorf("parent rule returned a leaf queue: %s", parentName)
			}
		}
		// the parent is set from the rule otherwise set it to the root
		if parentName == "" {
			parentName = configs.RootQueue
		}
		queueName = parentName + configs.DOT + childQueueName
	}
	// Log the result before we check the create flag
	log.Log(log.SchedApplication).Info("Tag rule intermediate result",
		zap.String("application", app.ApplicationID),
		zap.String("queue", queueName))
	// get the queue object
	queue := queueFn(queueName)
	// if we cannot create the queue it must exist, rule does not match otherwise
	if !tr.create && queue == nil {
		return "", nil
	}
	log.Log(log.SchedApplication).Info("Tag rule application placed",
		zap.String("application", app.ApplicationID),
		zap.String("queue", queueName))
	return queueName, nil
}