func()

in pkg/ingress/controller/openstack/octavia.go [391:492]
82 lines of code
17 McCabe index (conditional complexity)

func (os *OpenStack) EnsurePoolMembers(deleted bool, poolName string, lbID string, listenerID string, nodePort *int, nodes []*apiv1.Node) (*string, error) {
	logger := log.WithFields(log.Fields{"lbID": lbID, "listenerID": listenerID, "poolName": poolName})

	if deleted {
		pool, err := openstackutil.GetPoolByName(os.Octavia, poolName, lbID)
		if err != nil {
			if err != openstackutil.ErrNotFound {
				return nil, fmt.Errorf("error getting pool %s: %v", poolName, err)
			}
			return nil, nil
		}

		// Delete the existing pool, members are deleted automatically
		err = pools.Delete(os.Octavia, pool.ID).ExtractErr()
		if err != nil && !cpoerrors.IsNotFound(err) {
			return nil, fmt.Errorf("error deleting pool %s: %v", pool.ID, err)
		}

		_, err = os.waitLoadbalancerActiveProvisioningStatus(lbID)
		if err != nil {
			return nil, fmt.Errorf("error waiting for loadbalancer %s to be active: %v", lbID, err)
		}

		return nil, nil
	}

	pool, err := openstackutil.GetPoolByName(os.Octavia, poolName, lbID)
	if err != nil {
		if err != openstackutil.ErrNotFound {
			return nil, fmt.Errorf("error getting pool %s: %v", poolName, err)
		}

		logger.Info("creating pool")

		// Create new pool
		var opts pools.CreateOptsBuilder
		if listenerID != "" {
			opts = pools.CreateOpts{
				Name:        poolName,
				Protocol:    "HTTP",
				LBMethod:    pools.LBMethodRoundRobin,
				ListenerID:  listenerID,
				Persistence: nil,
			}
		} else {
			opts = pools.CreateOpts{
				Name:           poolName,
				Protocol:       "HTTP",
				LBMethod:       pools.LBMethodRoundRobin,
				LoadbalancerID: lbID,
				Persistence:    nil,
			}
		}
		pool, err = pools.Create(os.Octavia, opts).Extract()
		if err != nil {
			return nil, fmt.Errorf("error creating pool: %v", err)
		}

		logger.Info("pool created")

	}

	_, err = os.waitLoadbalancerActiveProvisioningStatus(lbID)
	if err != nil {
		return nil, fmt.Errorf("error waiting for loadbalancer %s to be active: %v", lbID, err)
	}

	// Batch update pool members
	var members []pools.BatchUpdateMemberOpts
	for _, node := range nodes {
		addr, err := getNodeAddressForLB(node)
		if err != nil {
			// Node failure, do not create member
			logger.WithFields(log.Fields{"nodeName": node.Name, "error": err}).Warn("failed to create LB pool member for node")
			continue
		}

		nodeName := node.Name
		member := pools.BatchUpdateMemberOpts{
			Name:         &nodeName,
			Address:      addr,
			ProtocolPort: *nodePort,
		}
		members = append(members, member)
	}
	// only allow >= 1 members or it will lead to openstack octavia issue
	if len(members) == 0 {
		return nil, fmt.Errorf("error because no members in pool: %s", pool.ID)
	}

	if err := pools.BatchUpdateMembers(os.Octavia, pool.ID, members).ExtractErr(); err != nil {
		return nil, fmt.Errorf("error batch updating members for pool %s: %v", pool.ID, err)
	}
	_, err = os.waitLoadbalancerActiveProvisioningStatus(lbID)
	if err != nil {
		return nil, fmt.Errorf("error waiting for loadbalancer %s to be active: %v", lbID, err)
	}

	logger.Info("pool members updated")

	return &pool.ID, nil
}