Merge pull request #224 from OrangeBao/feature_node

feat: add address for knode

pkg/clustertree/cluster-manager/cluster_controller.go

@@ -324,6 +324,46 @@ func (c *ClusterController) setupStorageControllers(mgr manager.Manager, nodes [
  return nil
 }
 
+func (c *ClusterController) setNodeStatus(ctx context.Context, nodeName string, leafClient kubernetes.Interface, node *corev1.Node, isNode2Node bool) error {
+ if isNode2Node {
+ if leafnode, err := leafClient.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{}); err != nil {
+ klog.Errorf("create node %s failed, cannot get node from leaf cluster, err: %v", nodeName, err)
+ return err
+ } else {
+ node.Status = leafnode.Status
+ address, err := leafUtils.SortAddress(ctx, c.RootClient, nodeName, leafClient, node.Status.Addresses)
+ if err != nil {
+ return err
+ }
+ node.Status.Addresses = address
+ return nil
+ }
+ }
+
+ leafnodes, err := leafClient.CoreV1().Nodes().List(ctx, metav1.ListOptions{
+ // TODO: LabelSelector
+ })
+ if err != nil {
+ klog.Errorf("create node %s failed, cannot get node from leaf cluster, err: %v", nodeName, err)
+ return err
+ }
+
+ if len(leafnodes.Items) == 0 {
+ klog.Errorf("create node %s failed, cannot get node from leaf cluster, len of leafnodes is 0", nodeName)
+ return err
+ }
+
+ address, err := leafUtils.SortAddress(ctx, c.RootClient, nodeName, leafClient, leafnodes.Items[0].Status.Addresses)
+
+ if err != nil {
+ return err
+ }
+
+ node.Status.Addresses = address
+
+ return nil
+}
+
 func (c *ClusterController) createNode(ctx context.Context, cluster *kosmosv1alpha1.Cluster, leafClient kubernetes.Interface) ([]*corev1.Node, error) {
  getNodeLen := func(cluster *kosmosv1alpha1.Cluster) int32 {
  if cluster.Spec.ClusterTreeOptions.Enable {
@@ -351,12 +391,18 @@ func (c *ClusterController) createNode(ctx context.Context, cluster *kosmosv1alp
  nodeAnnotations[utils.KosmosNodeOwnedByClusterAnnotations] = clusterName
  node.SetAnnotations(nodeAnnotations)
  }
+
+ if err := c.setNodeStatus(ctx, nodeName, leafClient, node, isNode2Node); err != nil {
+ return nil, err
+ }
+
  node.Status.NodeInfo.KubeletVersion = serverVersion.GitVersion
  node.Status.DaemonEndpoints = corev1.NodeDaemonEndpoints{
  KubeletEndpoint: corev1.DaemonEndpoint{
  Port: c.Options.ListenPort,
  },
  }
+
  node, err = c.RootClient.CoreV1().Nodes().Create(ctx, node, metav1.CreateOptions{})
  if err != nil {
  if !errors.IsAlreadyExists(err) {

pkg/clustertree/cluster-manager/controllers/node_resources_controller.go

@@ -128,7 +128,7 @@ func (c *NodeResourcesController) Reconcile(ctx context.Context, request reconci
  }, err
  }
  } else {
- leafNodeName := fmt.Sprintf("%s%s", utils.KosmosNodePrefix, c.Cluster.Name)
+ leafNodeName := c.Cluster.Name
  leafResource, err := c.GlobalLeafManager.GetLeafResource(leafNodeName)
  if err != nil {
  klog.Errorf("Could not get leafResource,Error: %v", err)
@@ -174,6 +174,23 @@ func (c *NodeResourcesController) Reconcile(ctx context.Context, request reconci
  clone.Labels = mergeMap(rootNode.GetLabels(), node.GetLabels())
  clone.Annotations = mergeMap(rootNode.GetAnnotations(), node.GetAnnotations())
  clone.Status = node.Status
+ // TODO: @duanmengkk
+ leafNodeName := c.Cluster.Name
+ leafResource, err := c.GlobalLeafManager.GetLeafResource(leafNodeName)
+ if err != nil {
+ klog.Errorf("Could not get leafResource,Error: %v", err)
+ return controllerruntime.Result{
+ RequeueAfter: RequeueTime,
+ }, err
+ }
+ address, err := leafUtils.SortAddress(ctx, c.RootClientset, rootNode.Name, leafResource.Clientset, node.Status.Addresses)
+ if err != nil {
+ return controllerruntime.Result{
+ RequeueAfter: RequeueTime,
+ }, err
+ }
+
+ node.Status.Addresses = address
  }
  }
 

pkg/clustertree/cluster-manager/utils/rootcluster.go

@@ -1,6 +1,17 @@
 package leafUtils
 
-import kosmosv1alpha1 "github.com/kosmos.io/kosmos/pkg/apis/kosmos/v1alpha1"
+import (
+ "context"
+ "sort"
+
+ corev1 "k8s.io/api/core/v1"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/client-go/kubernetes"
+ "k8s.io/klog"
+
+ kosmosv1alpha1 "github.com/kosmos.io/kosmos/pkg/apis/kosmos/v1alpha1"
+ "github.com/kosmos.io/kosmos/pkg/utils"
+)
 
 const (
  RootClusterAnnotationKey = "kosmos.io/cluster-role"
@@ -15,3 +26,56 @@ func IsRootCluster(cluster *kosmosv1alpha1.Cluster) bool {
  }
  return false
 }
+
+func SortAddress(ctx context.Context, rootClient kubernetes.Interface, nodeName string, leafClient kubernetes.Interface, originAddress []corev1.NodeAddress) ([]corev1.NodeAddress, error) {
+ rootnodes, err := rootClient.CoreV1().Nodes().List(ctx, metav1.ListOptions{})
+ if err != nil {
+ klog.Errorf("create node %s failed, cannot get node from root cluster, err: %v", nodeName, err)
+ return nil, err
+ }
+
+ if len(rootnodes.Items) == 0 {
+ klog.Errorf("create node %s failed, cannot get node from root cluster, len of leafnodes is 0", nodeName)
+ return nil, err
+ }
+
+ isIPv4First := true
+ for _, addr := range rootnodes.Items[0].Status.Addresses {
+ if addr.Type == corev1.NodeInternalIP {
+ if utils.IsIPv6(addr.Address) {
+ isIPv4First = false
+ }
+ break
+ }
+ }
+
+ address := []corev1.NodeAddress{}
+
+ for _, addr := range originAddress {
+ if addr.Type == corev1.NodeInternalIP {
+ address = append(address, corev1.NodeAddress{Type: corev1.NodeInternalIP, Address: addr.Address})
+ }
+ }
+
+ sort.Slice(address, func(i, j int) bool {
+ if isIPv4First {
+ if !utils.IsIPv6(address[i].Address) && utils.IsIPv6(address[j].Address) {
+ return true
+ }
+ if utils.IsIPv6(address[i].Address) && !utils.IsIPv6(address[j].Address) {
+ return false
+ }
+ return true
+ } else {
+ if !utils.IsIPv6(address[i].Address) && utils.IsIPv6(address[j].Address) {
+ return false
+ }
+ if utils.IsIPv6(address[i].Address) && !utils.IsIPv6(address[j].Address) {
+ return true
+ }
+ return true
+ }
+ })
+
+ return address, nil
+}