func()

in pkg/scheduler/placement/provided_rule.go [72:145]

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

func (pr *providedRule) placeApplication(app *objects.Application, queueFn func(string) *objects.Queue) (string, error) {
	// since this is the provided rule we must have a queue in the info already
	queueName := app.GetQueuePath()
	if queueName == "" {
		return "", nil
	}

	// before anything run the filter
	if !pr.filter.allowUser(app.GetUser()) {
		log.Log(log.SchedApplication).Debug("Provided rule filtered",
			zap.String("application", app.ApplicationID),
			zap.Any("user", app.GetUser()))
		return "", nil
	}
	var parentName string
	var err error

	// 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(queueName)
		if err = configs.IsQueueNameValid(childQueueName); err != nil {
			return "", err
		}
		// run the parent rule if set
		if pr.parent != nil {
			parentName, err = pr.parent.placeApplication(app, queueFn)
			// failed parent rule, fail this rule
			if err != nil {
				return "", err
			}
			// rule did not return a parent: 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
		}
		// Make it a fully qualified queue
		queueName = parentName + configs.DOT + childQueueName
	}
	// Log the result before we check the create flag
	log.Log(log.SchedApplication).Debug("Provided 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 must exist
	if !pr.create && queue == nil {
		return "", nil
	}
	log.Log(log.SchedApplication).Info("Provided rule application placed",
		zap.String("application", app.ApplicationID),
		zap.String("queue", queueName))
	return queueName, nil
}