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.

Sign up for GitHub

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

refactor: allow variants to implement their own k8s providerID parsing logic #938

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

refactor: allow variants to implement their own k8s providerID parsing logic #938

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
10 changes: 5 additions & 5 deletions pkg/controllers/tagging/tagging_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,17 +225,17 @@ func (tc *Controller) process() bool {
recordWorkItemLatencyMetrics(workItemDequeuingTimeWorkItemMetric, timeTaken)
klog.Infof("Dequeuing latency %f seconds", timeTaken)

instanceID, err := awsv1.KubernetesInstanceID(workItem.node.Spec.ProviderID).MapToAWSInstanceID()
instanceID, err := awsv1.ParseProviderID(workItem.node.Spec.ProviderID)
if err != nil {
err = fmt.Errorf("Error in getting instanceID for node %s, error: %v", workItem.node.GetName(), err)
utilruntime.HandleError(err)
return nil
}
klog.Infof("Instance ID of work item %s is %s", workItem, instanceID)

if variant.IsVariantNode(string(instanceID)) {
if variant.IsVariantNode(instanceID) {
klog.Infof("Skip processing the node %s since it is a %s node",
instanceID, variant.NodeType(string(instanceID)))
instanceID, variant.NodeType(instanceID))
tc.workqueue.Forget(obj)
return nil
}
Expand Down Expand Up @@ -297,7 +297,7 @@ func (tc *Controller) tagEc2Instance(node *v1.Node) error {
return nil
}

instanceID, _ := awsv1.KubernetesInstanceID(node.Spec.ProviderID).MapToAWSInstanceID()
instanceID, _ := awsv1.ParseProviderID(node.Spec.ProviderID)

err := tc.cloud.TagResource(string(instanceID), tc.tags)

Expand Down Expand Up @@ -349,7 +349,7 @@ func (tc *Controller) untagNodeResources(node *v1.Node) error {
// untagEc2Instances deletes the provided tags to each EC2 instances in
// the cluster.
func (tc *Controller) untagEc2Instance(node *v1.Node) error {
instanceID, _ := awsv1.KubernetesInstanceID(node.Spec.ProviderID).MapToAWSInstanceID()
instanceID, _ := awsv1.ParseProviderID(node.Spec.ProviderID)

err := tc.cloud.UntagResource(string(instanceID), tc.tags)

Expand Down
53 changes: 27 additions & 26 deletions pkg/providers/v1/aws.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
"context"
"fmt"
"io"
"k8s.io/cloud-provider-aws/pkg/providers/v1/awsnode"
Copy link
Member

Choose a reason for hiding this comment

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

nit: can we group the import correctly ?

"net"
"regexp"
"sort"
Expand Down Expand Up @@ -418,7 +419,7 @@ func InstanceIDIndexFunc(obj interface{}) ([]string, error) {
// provider ID hasn't been populated yet
return []string{""}, nil
}
instanceID, err := KubernetesInstanceID(node.Spec.ProviderID).MapToAWSInstanceID()
instanceID, err := ParseProviderID(node.Spec.ProviderID)
if err != nil {
//logging the error as warning as Informer.AddIndexers would panic if there is an error
klog.Warningf("error mapping node %q's provider ID %q to instance ID: %v", node.Name, node.Spec.ProviderID, err)
Expand Down Expand Up @@ -832,16 +833,16 @@ func extractIPv6NodeAddresses(instance *ec2.Instance) ([]v1.NodeAddress, error)
// This method will not be called from the node that is requesting this ID. i.e. metadata service
// and other local methods cannot be used here
func (c *Cloud) NodeAddressesByProviderID(ctx context.Context, providerID string) ([]v1.NodeAddress, error) {
instanceID, err := KubernetesInstanceID(providerID).MapToAWSInstanceID()
instanceID, err := ParseProviderID(providerID)
if err != nil {
return nil, err
}

if v := variant.GetVariant(string(instanceID)); v != nil {
return v.NodeAddresses(string(instanceID), c.vpcID)
if v := variant.GetVariant(instanceID); v != nil {
return v.NodeAddresses(instanceID, c.vpcID)
}

instance, err := describeInstance(c.ec2, instanceID)
instance, err := describeInstance(c.ec2, string(instanceID))
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -871,17 +872,17 @@ func (c *Cloud) NodeAddressesByProviderID(ctx context.Context, providerID string
// InstanceExistsByProviderID returns true if the instance with the given provider id still exists.
// If false is returned with no error, the instance will be immediately deleted by the cloud controller manager.
func (c *Cloud) InstanceExistsByProviderID(ctx context.Context, providerID string) (bool, error) {
instanceID, err := KubernetesInstanceID(providerID).MapToAWSInstanceID()
instanceID, err := ParseProviderID(providerID)
if err != nil {
return false, err
}

if v := variant.GetVariant(string(instanceID)); v != nil {
return v.InstanceExists(string(instanceID), c.vpcID)
if v := variant.GetVariant(instanceID); v != nil {
return v.InstanceExists(instanceID, c.vpcID)
}

request := &ec2.DescribeInstancesInput{
InstanceIds: []*string{instanceID.awsString()},
InstanceIds: []*string{instanceID.AwsString()},
}

instances, err := c.ec2.DescribeInstances(request)
Expand Down Expand Up @@ -910,17 +911,17 @@ func (c *Cloud) InstanceExistsByProviderID(ctx context.Context, providerID strin

// InstanceShutdownByProviderID returns true if the instance is terminated
func (c *Cloud) InstanceShutdownByProviderID(ctx context.Context, providerID string) (bool, error) {
instanceID, err := KubernetesInstanceID(providerID).MapToAWSInstanceID()
instanceID, err := ParseProviderID(providerID)
if err != nil {
return false, err
}

if v := variant.GetVariant(string(instanceID)); v != nil {
return v.InstanceShutdown(string(instanceID), c.vpcID)
if v := variant.GetVariant(instanceID); v != nil {
return v.InstanceShutdown(instanceID, c.vpcID)
}

request := &ec2.DescribeInstancesInput{
InstanceIds: []*string{instanceID.awsString()},
InstanceIds: []*string{instanceID.AwsString()},
}

instances, err := c.ec2.DescribeInstances(request)
Expand Down Expand Up @@ -969,16 +970,16 @@ func (c *Cloud) InstanceID(ctx context.Context, nodeName types.NodeName) (string
// This method will not be called from the node that is requesting this ID. i.e. metadata service
// and other local methods cannot be used here
func (c *Cloud) InstanceTypeByProviderID(ctx context.Context, providerID string) (string, error) {
instanceID, err := KubernetesInstanceID(providerID).MapToAWSInstanceID()
instanceID, err := ParseProviderID(providerID)
if err != nil {
return "", err
}

if v := variant.GetVariant(string(instanceID)); v != nil {
return v.InstanceTypeByProviderID(string(instanceID))
if v := variant.GetVariant(instanceID); v != nil {
return v.InstanceTypeByProviderID(instanceID)
}

instance, err := describeInstance(c.ec2, instanceID)
instance, err := describeInstance(c.ec2, string(instanceID))
if err != nil {
return "", err
}
Expand Down Expand Up @@ -1010,13 +1011,13 @@ func (c *Cloud) GetZone(ctx context.Context) (cloudprovider.Zone, error) {
// This is particularly useful in external cloud providers where the kubelet
// does not initialize node data.
func (c *Cloud) GetZoneByProviderID(ctx context.Context, providerID string) (cloudprovider.Zone, error) {
instanceID, err := KubernetesInstanceID(providerID).MapToAWSInstanceID()
instanceID, err := ParseProviderID(providerID)
if err != nil {
return cloudprovider.Zone{}, err
}

if v := variant.GetVariant(string(instanceID)); v != nil {
return v.GetZone(string(instanceID), c.vpcID, c.region)
if v := variant.GetVariant(instanceID); v != nil {
return v.GetZone(instanceID, c.vpcID, c.region)
}

instance, err := c.getInstanceByID(string(instanceID))
Expand Down Expand Up @@ -2651,7 +2652,7 @@ func (c *Cloud) getTaggedSecurityGroups() (map[string]*ec2.SecurityGroup, error)

// Open security group ingress rules on the instances so that the load balancer can talk to them
// Will also remove any security groups ingress rules for the load balancer that are _not_ needed for allInstances
func (c *Cloud) updateInstanceSecurityGroupsForLoadBalancer(lb *elb.LoadBalancerDescription, instances map[InstanceID]*ec2.Instance, annotations map[string]string) error {
func (c *Cloud) updateInstanceSecurityGroupsForLoadBalancer(lb *elb.LoadBalancerDescription, instances map[awsnode.NodeID]*ec2.Instance, annotations map[string]string) error {
if c.cfg.Global.DisableSecurityGroupIngress {
return nil
}
Expand Down Expand Up @@ -3228,15 +3229,15 @@ func nodeNameToIPAddress(nodeName string) string {
return strings.ReplaceAll(nodeName, "-", ".")
}

func (c *Cloud) nodeNameToInstanceID(nodeName types.NodeName) (InstanceID, error) {
func (c *Cloud) nodeNameToInstanceID(nodeName types.NodeName) (awsnode.NodeID, error) {
if strings.HasPrefix(string(nodeName), rbnNamePrefix) {
// depending on if you use a RHEL (e.g. AL2) or Debian (e.g. standard Ubuntu) based distribution, the
// hostname on the machine may be either i-00000000000000001 or i-00000000000000001.region.compute.internal.
// This handles both scenarios by returning anything before the first '.' in the node name if it has an RBN prefix.
if idx := strings.IndexByte(string(nodeName), '.'); idx != -1 {
return InstanceID(nodeName[0:idx]), nil
return awsnode.NodeID(nodeName[0:idx]), nil
}
return InstanceID(nodeName), nil
return awsnode.NodeID(nodeName), nil
}
if len(nodeName) == 0 {
return "", fmt.Errorf("no nodeName provided")
Expand All @@ -3254,10 +3255,10 @@ func (c *Cloud) nodeNameToInstanceID(nodeName types.NodeName) (InstanceID, error
return "", fmt.Errorf("node has no providerID")
}

return KubernetesInstanceID(node.Spec.ProviderID).MapToAWSInstanceID()
return ParseProviderID(node.Spec.ProviderID)
}

func (c *Cloud) instanceIDToNodeName(instanceID InstanceID) (types.NodeName, error) {
func (c *Cloud) instanceIDToNodeName(instanceID awsnode.NodeID) (types.NodeName, error) {
if len(instanceID) == 0 {
return "", fmt.Errorf("no instanceID provided")
}
Expand Down
3 changes: 1 addition & 2 deletions pkg/providers/v1/aws_instance.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"k8s.io/apimachinery/pkg/types"

"k8s.io/cloud-provider-aws/pkg/providers/v1/iface"
)

Expand Down Expand Up @@ -67,5 +66,5 @@ func newAWSInstance(ec2Service iface.EC2, instance *ec2.Instance) *awsInstance {

// Gets the full information about this instance from the EC2 API
func (i *awsInstance) describeInstance() (*ec2.Instance, error) {
return describeInstance(i.ec2, InstanceID(i.awsID))
return describeInstance(i.ec2, i.awsID)
}
7 changes: 4 additions & 3 deletions pkg/providers/v1/aws_loadbalancer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
"crypto/sha1"
"encoding/hex"
"fmt"
"k8s.io/cloud-provider-aws/pkg/providers/v1/awsnode"
"reflect"
"regexp"
"strconv"
Expand Down Expand Up @@ -781,7 +782,7 @@ func (c *Cloud) chunkTargetDescriptions(targets []*elbv2.TargetDescription, chun

// updateInstanceSecurityGroupsForNLB will adjust securityGroup's settings to allow inbound traffic into instances from clientCIDRs and portMappings.
// TIP: if either instances or clientCIDRs or portMappings are nil, then the securityGroup rules for lbName are cleared.
func (c *Cloud) updateInstanceSecurityGroupsForNLB(lbName string, instances map[InstanceID]*ec2.Instance, subnetCIDRs []string, clientCIDRs []string, portMappings []nlbPortMapping) error {
func (c *Cloud) updateInstanceSecurityGroupsForNLB(lbName string, instances map[awsnode.NodeID]*ec2.Instance, subnetCIDRs []string, clientCIDRs []string, portMappings []nlbPortMapping) error {
if c.cfg.Global.DisableSecurityGroupIngress {
return nil
}
Expand Down Expand Up @@ -1430,7 +1431,7 @@ func (c *Cloud) ensureLoadBalancerHealthCheck(loadBalancer *elb.LoadBalancerDesc
}

// Makes sure that exactly the specified hosts are registered as instances with the load balancer
func (c *Cloud) ensureLoadBalancerInstances(loadBalancerName string, lbInstances []*elb.Instance, instanceIDs map[InstanceID]*ec2.Instance) error {
func (c *Cloud) ensureLoadBalancerInstances(loadBalancerName string, lbInstances []*elb.Instance, instanceIDs map[awsnode.NodeID]*ec2.Instance) error {
expected := sets.NewString()
for id := range instanceIDs {
expected.Insert(string(id))
Expand Down Expand Up @@ -1607,7 +1608,7 @@ func proxyProtocolEnabled(backend *elb.BackendServerDescription) bool {
// findInstancesForELB gets the EC2 instances corresponding to the Nodes, for setting up an ELB
// We ignore Nodes (with a log message) where the instanceid cannot be determined from the provider,
// and we ignore instances which are not found
func (c *Cloud) findInstancesForELB(nodes []*v1.Node, annotations map[string]string) (map[InstanceID]*ec2.Instance, error) {
func (c *Cloud) findInstancesForELB(nodes []*v1.Node, annotations map[string]string) (map[awsnode.NodeID]*ec2.Instance, error) {

targetNodes := filterTargetNodes(nodes, annotations)

Expand Down
7 changes: 4 additions & 3 deletions pkg/providers/v1/aws_loadbalancer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ package aws

import (
"fmt"
"k8s.io/cloud-provider-aws/pkg/providers/v1/awsnode"
"reflect"
"testing"
"time"
Expand Down Expand Up @@ -592,7 +593,7 @@ func TestCloud_findInstancesForELB(t *testing.T) {
return
}

want := map[InstanceID]*ec2.Instance{
want := map[awsnode.NodeID]*ec2.Instance{
"i-self": awsServices.selfInstance,
}
got, err := c.findInstancesForELB([]*v1.Node{defaultNode}, nil)
Expand All @@ -601,9 +602,9 @@ func TestCloud_findInstancesForELB(t *testing.T) {

// Add a new EC2 instance
awsServices.instances = append(awsServices.instances, newInstance)
want = map[InstanceID]*ec2.Instance{
want = map[awsnode.NodeID]*ec2.Instance{
"i-self": awsServices.selfInstance,
InstanceID(aws.StringValue(newInstance.InstanceId)): newInstance,
awsnode.NodeID(aws.StringValue(newInstance.InstanceId)): newInstance,
}
got, err = c.findInstancesForELB([]*v1.Node{defaultNode, newNode}, nil)
assert.NoError(t, err)
Expand Down
4 changes: 2 additions & 2 deletions pkg/providers/v1/aws_routes.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,9 @@ package aws
import (
"context"
"fmt"

"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"k8s.io/cloud-provider-aws/pkg/providers/v1/awsnode"
"k8s.io/klog/v2"

cloudprovider "k8s.io/cloud-provider"
Expand Down Expand Up @@ -114,7 +114,7 @@ func (c *Cloud) ListRoutes(ctx context.Context, clusterName string) ([]*cloudpro
if instanceID != "" {
_, found := instances[instanceID]
if found {
node, err := c.instanceIDToNodeName(InstanceID(instanceID))
node, err := c.instanceIDToNodeName(awsnode.NodeID(instanceID))
if err != nil {
return nil, err
}
Expand Down
3 changes: 2 additions & 1 deletion pkg/providers/v1/aws_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import (
"errors"
"fmt"
"io"
"k8s.io/cloud-provider-aws/pkg/providers/v1/awsnode"
"math/rand"
"reflect"
"sort"
Expand Down Expand Up @@ -2399,7 +2400,7 @@ func TestNodeNameToInstanceID(t *testing.T) {
func TestInstanceIDToNodeName(t *testing.T) {
testCases := []struct {
name string
instanceID InstanceID
instanceID awsnode.NodeID
node *v1.Node
expectedNodeName types.NodeName
expectedErr error
Expand Down
11 changes: 11 additions & 0 deletions pkg/providers/v1/awsnode/identifier.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
package awsnode

import "github.com/aws/aws-sdk-go/aws"

// NodeID is the ID used to uniquely identify a node within an AWS service
type NodeID string
Copy link
Contributor

Choose a reason for hiding this comment

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

I think this could be more useful as an interface, we have too many string alias types in here already 😛

Something like:

type AWSProviderID interface {
    ComputeID() string
}

Then we can implement an EC2ProviderID, FargateProviderID, etc.

The func-s in the Variant interface can accept an AWSProviderID and cast to the relevant type as necessary, providerID.(EC2ProviderID)

Copy link
Author

Choose a reason for hiding this comment

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

hmm I actually am not quite following the need for an interface here. While the NodeID might have different formats, I dont see why there would be different implementations required for those different formats. Along that same line, while it sounds potentially useful to have strongly typed NodeID's, there is currently not a need for it — the casting you mention wouldnt actually be necessary anywhere. That to me signals that an interface may be overkill, but perhaps Im not thinking about a future use-case where having an interface would be ideal.

So as it stands now, I actually think the ergonomics of the string alias type is fits the needs quite well. Let me know your thoughts.

Copy link
Author

Choose a reason for hiding this comment

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

also, much less of a concern, but moving to an interface would substantially increase the size of the changes in this commit.


// AwsString returns a pointer to the string value of the NodeID. Useful for AWS APIs
func (i NodeID) AwsString() *string {
return aws.String(string(i))
}
Loading