Merge pull request #243 from Zippo-Wang/feat_security_group_id

feat(securityGroup): add security group of cloud config

docs/create-cloud-config-secret.md

@@ -14,6 +14,7 @@ project-id=62fac0038c52b5eb0970dd9c1a0026a
 # The VPC where your cluster resides
 id=0b876fcb-6a0b-47a3-8067-80fe9245a3da
 subnet-id=d2aa75cc-e356-4dc3-abb0-87078923a168
+security-group-id=48288809-1234-5678-abcd-eb30b2d16cd4
 
 EOF
 

docs/huawei-cloud-controller-manager-configuration.md

@@ -25,6 +25,7 @@ auth-url=
 [Vpc]
 id=
 subnet-id=
+security-group-id=
 ```
 
 > After modification, CCM needs to be restarted to load the data.
@@ -37,7 +38,7 @@ This section provides Huawei Cloud IAM configuration and authentication informat
 
 * `region` Required. This is the Huawei Cloud region.
 
-  **Note**: The `region` must be the same as the ECSes of the Kubernetes cluster.
+  **Note**: The `region` must be the same as the ECS of the Kubernetes cluster.
 
 * `access-key` Required. The access key of the Huawei Cloud.
 
@@ -46,7 +47,7 @@ This section provides Huawei Cloud IAM configuration and authentication informat
 * `project-id` Optional. The Project ID of the Huawei Cloud. 
   See [Obtaining a Project ID](https://support.huaweicloud.com/intl/en-us/api-evs/evs_04_0046.html).
 
-  **Note**: The `project-id` must be the same as the ECSes of the Kubernetes cluster.
+  **Note**: The `project-id` must be the same as the ECS of the Kubernetes cluster.
 
 * `cloud` Optional. The endpoint of the cloud provider. Defaults to `myhuaweicloud.com`'`.
 
@@ -56,9 +57,13 @@ This section provides Huawei Cloud IAM configuration and authentication informat
 
 This section contains network configuration information.
 
-* `id` Optional. Specifies the VPC used by ECSes of the Kubernetes cluster.
+* `id` Optional. Specifies the VPC used by ECS of the Kubernetes cluster.
 
-* `subnet-id` Optional. Specifies the IPv4 subnet ID used by ECSes of the Kubernetes cluster.
+* `subnet-id` Optional. Specifies the IPv4 subnet ID used by ECS of the Kubernetes cluster.
+
+* `security-group-id` Optional. A security group ID used for associating with nodes.
+  When new nodes are added to the cluster, they will be automatically associated with the security group.
+  Conversely, when nodes are removed, the association will be automatically removed as well.
 
 ## Loadbalancer Configuration
 

manifests/cloud-config

@@ -10,3 +10,4 @@ auth-url=<optional, the cloud IAM domain name>
 [Vpc]
 id=<your vpc ID>
 subnet-id=<your IPV4 subnet ID>
+security-group-id=<optional, your security group ID>

pkg/cloudprovider/huaweicloud/huaweicloud.go

@@ -767,13 +767,23 @@ func (h *CloudProvider) listenerDeploy() error {
 		invalidServiceCache: gocache.New(5*time.Minute, 10*time.Minute),
 	}
 
+	secListener := &SecurityGroupListener{
+		kubeClient:      h.kubeClient,
+		ecsClient:       h.ecsClient,
+		securityGroupID: h.cloudConfig.VpcOpts.SecurityGroupID,
+
+		stopChannel: make(chan struct{}, 1),
+	}
+
 	clusterName := h.cloudControllerManagerOpts.KubeCloudShared.ClusterName
 	id, err := os.Hostname()
 	if err != nil {
 		return fmt.Errorf("failed to elect leader in listener EndpointSlice : %s", err)
 	}
 
 	go leaderElection(id, h.restConfig, h.eventRecorder, func(ctx context.Context) {
+		go secListener.startSecurityGroupListener()
+
 		listener.startEndpointListener(func(service *v1.Service, isDelete bool) {
 			klog.Infof("Got service %s/%s using loadbalancer class %s",
 				service.Namespace, service.Name, utils.ToString(service.Spec.LoadBalancerClass))
@@ -832,6 +842,7 @@ func (h *CloudProvider) listenerDeploy() error {
 	}, func() {
 		listener.goroutinePool.Stop()
 		listener.stopListenerSlice()
+		secListener.stopSecurityGroupListener()
 	})
 
 	return nil

pkg/cloudprovider/huaweicloud/security_group_listener.go

@@ -0,0 +1,124 @@
+package huaweicloud
+
+import (
+	"context"
+	"time"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"sigs.k8s.io/cloud-provider-huaweicloud/pkg/cloudprovider/huaweicloud/wrapper"
+
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/klog/v2"
+)
+
+type SecurityGroupListener struct {
+	kubeClient *corev1.CoreV1Client
+	ecsClient  *wrapper.EcsClient
+
+	securityGroupID string
+
+	stopChannel chan struct{}
+}
+
+func (s *SecurityGroupListener) startSecurityGroupListener() {
+	if s.securityGroupID == "" {
+		klog.Infof(`"security-group-id" is empty, nodes are added or removed will not be associated or disassociated 
+	any security groups`)
+		return
+	}
+
+	klog.Info("starting SecurityGroupListener")
+	for {
+		securityGroupInformer, err := s.CreateSecurityGroupInformer()
+		if err != nil {
+			klog.Errorf("failed to watch kubernetes cluster node list, starting SecurityGroupListener failed: %s", err)
+			continue
+		}
+
+		go securityGroupInformer.Run(s.stopChannel)
+		break
+	}
+}
+
+func (s *SecurityGroupListener) CreateSecurityGroupInformer() (cache.SharedIndexInformer, error) {
+	nodeList, err := s.kubeClient.Nodes().List(context.TODO(), metav1.ListOptions{})
+	if err != nil {
+		klog.Errorf("failed to query a list of node, try again later, error: %s", err)
+		time.Sleep(5 * time.Second)
+		return nil, err
+	}
+	nodeInformer := cache.NewSharedIndexInformer(
+		&cache.ListWatch{
+			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
+				if options.ResourceVersion == "" || options.ResourceVersion == "0" {
+					options.ResourceVersion = nodeList.ResourceVersion
+				}
+				return s.kubeClient.Nodes().List(context.TODO(), options)
+			},
+			WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
+				if options.ResourceVersion == "" || options.ResourceVersion == "0" {
+					options.ResourceVersion = nodeList.ResourceVersion
+				}
+				return s.kubeClient.Nodes().Watch(context.TODO(), options)
+			},
+		},
+		&v1.Node{},
+		0,
+		cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
+	)
+
+	_, err = nodeInformer.AddEventHandlerWithResyncPeriod(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			kubeNode, ok := obj.(*v1.Node)
+			if !ok {
+				klog.Errorf("detected that a node has been added, but the type conversion failed: %#v", obj)
+				return
+			}
+			klog.Infof("detected that a new node has been added to the cluster(%v): %s", ok, kubeNode.Name)
+
+			ecsNode, err := s.ecsClient.GetByNodeName(kubeNode.Name)
+			if err != nil {
+				klog.Error("Add node: can not get kubernetes node name: %v", err)
+				return
+			}
+			err = s.ecsClient.AssociateSecurityGroup(ecsNode.Id, s.securityGroupID)
+			if err != nil {
+				klog.Errorf("failed to associate security group %s to ECS %s: %s", s.securityGroupID, ecsNode.Id, err)
+			}
+		},
+		UpdateFunc: func(oldObj, newObj interface{}) {},
+		DeleteFunc: func(obj interface{}) {
+			kubeNode, ok := obj.(*v1.Node)
+			if !ok {
+				klog.Errorf("detected that a node has been removed, but the type conversion failed: %#v", obj)
+				return
+			}
+			klog.Infof("detected that a node has been removed to the cluster: %s", kubeNode.Name)
+
+			ecsNode, err := s.ecsClient.GetByNodeName(kubeNode.Name)
+			if err != nil {
+				klog.Error("Delete node: can not get kubernetes node name: %v", err)
+				return
+			}
+			err = s.ecsClient.DisassociateSecurityGroup(ecsNode.Id, s.securityGroupID)
+			if err != nil {
+				klog.Errorf("failed to disassociate security group %s from ECS %s: %s", s.securityGroupID, ecsNode.Id, err)
+			}
+		},
+	}, 5*time.Second)
+	if err != nil {
+		klog.Errorf("failed to start nodeEventHandler, try again later, error: %s", err)
+		time.Sleep(5 * time.Second)
+		return nil, err
+	}
+	return nodeInformer, nil
+}
+
+func (s *SecurityGroupListener) stopSecurityGroupListener() {
+	klog.Warningf("Stop listening to Security Group")
+	s.stopChannel <- struct{}{}
+}

pkg/cloudprovider/huaweicloud/wrapper/ecs.go

@@ -287,6 +287,44 @@ func (e *EcsClient) ListSecurityGroups(instanceID string) ([]model.NovaSecurityG
 	return rst, err
 }
 
+func (e *EcsClient) AssociateSecurityGroup(instanceID, securityGroupID string) error {
+	return e.wrapper(func(c *ecs.EcsClient) (interface{}, error) {
+		return c.NovaAssociateSecurityGroup(&model.NovaAssociateSecurityGroupRequest{
+			ServerId: instanceID,
+			Body: &model.NovaAssociateSecurityGroupRequestBody{
+				AddSecurityGroup: &model.NovaAddSecurityGroupOption{
+					Name: securityGroupID,
+				},
+			},
+		})
+	})
+}
+
+func (e *EcsClient) DisassociateSecurityGroup(instanceID, securityGroupID string) error {
+	err := e.wrapper(func(c *ecs.EcsClient) (interface{}, error) {
+		return c.NovaDisassociateSecurityGroup(&model.NovaDisassociateSecurityGroupRequest{
+			ServerId: instanceID,
+			Body: &model.NovaDisassociateSecurityGroupRequestBody{
+				RemoveSecurityGroup: &model.NovaRemoveSecurityGroupOption{
+					Name: securityGroupID,
+				},
+			},
+		})
+	})
+
+	if err != nil {
+		notAssociated := "not associated with the instance"
+		notFound := "is not found for project"
+		if strings.Contains(err.Error(), notAssociated) || strings.Contains(err.Error(), notFound) {
+			klog.Errorf("failed to disassociate security group %v from instance %v: %v",
+				securityGroupID, instanceID, err)
+			return nil
+		}
+	}
+
+	return err
+}
+
 // addToNodeAddresses appends the NodeAddresses to the passed-by-pointer slice, only if they do not already exist.
 func addToNodeAddresses(addresses *[]v1.NodeAddress, addAddresses ...v1.NodeAddress) {
 	for _, add := range addAddresses {

pkg/config/config.go

@@ -40,8 +40,9 @@ type CloudConfig struct {
 }
 
 type VpcOptions struct {
-	ID       string `gcfg:"id"`
-	SubnetID string `gcfg:"subnet-id"`
+	ID              string `gcfg:"id"`
+	SubnetID        string `gcfg:"subnet-id"`
+	SecurityGroupID string `gcfg:"security-group-id"`
 }
 
 type AuthOptions struct {