HyperNode Controller 深度解析

// pkg/controllers/hypernode/hypernode_controller.go
func init() {
    framework.RegisterController(&hyperNodeController{})
}

const (
    name = "hyperNode-controller"
)

// pkg/controllers/hypernode/hypernode_controller.go
func (hn *hyperNodeController) Run(stopCh <-chan struct{}) {
    hn.vcInformerFactory.Start(stopCh)
    hn.informerFactory.Start(stopCh)
    // 等待缓存同步...
    hn.discoveryManager.Start()
    go hn.watchDiscoveryResults()
    go hn.processHyperNodeQueue()
    <-stopCh
    hn.discoveryManager.Stop()
    hn.hyperNodeQueue.ShutDown()
}

type Manager interface {
    Start() error
    Stop()
    ResultChannel() <-chan Result
    ResultSynced(source string)
}

type Result struct {
    HyperNodes []*topologyv1alpha1.HyperNode
    Source     string
}

networkTopologyDiscovery:
  - source: label
    enabled: true
    config:
      networkTopologyTypes:
        topologyA2:
          - nodeLabel: "volcano.sh/tor"
          - nodeLabel: "kubernetes.io/hostname"
        topologyA3:
          - nodeLabel: "volcano.sh/hypercluster"
          - nodeLabel: "volcano.sh/hypernode"
          - nodeLabel: "kubernetes.io/hostname"

func (l *labelDiscoverer) buildHyperNodeName(topologyTypeName, key, value string, tier int,
    hyperNodeInfoMap map[string]HyperNodeInfo) (string, error)

const ufmPortsPath = "/ufmRest/resources/ports"

func (u *ufmDiscoverer) getCredentialsFromSecret(name, namespace string) (string, string, error) {
    secret, err := u.kubeClient.CoreV1().Secrets(namespace).Get(
        context.TODO(), name, metav1.GetOptions{})
    // 从 secret.Data 中提取 "username" 和 "password"
}

type UFMInterface struct {
    Description     string `json:"description"`
    Tier            int    `json:"tier"`
    SystemName      string `json:"system_name"`
    PeerNodeName    string `json:"peer_node_name"`
    NodeDescription string `json:"node_description"`
}

func (u *ufmDiscoverer) periodicDiscovery(outputCh chan []*topologyv1alpha1.HyperNode) {
    u.discoverAndSend(outputCh)  // 立即执行一次
    ticker := time.NewTicker(u.discoveryInterval)
    defer ticker.Stop()
    for {
        select {
        case <-ticker.C:
            u.discoverAndSend(outputCh)
        case <-u.stopCh:
            close(outputCh)
            return
        }
    }
}

type Discoverer interface {
    Start() (chan []*topologyv1alpha1.HyperNode, error)
    Stop() error
    Name() string
    ResultSynced()
}

type DiscovererConstructor func(cfg DiscoveryConfig, kubeClient clientset.Interface,
    vcClient vcclientset.Interface) Discoverer

var discovererRegistry = make(map[string]DiscovererConstructor)

func RegisterDiscoverer(source string, constructor DiscovererConstructor) {
    mutex.Lock()
    defer mutex.Unlock()
    discovererRegistry[source] = constructor
}

func NewDiscoverer(cfg DiscoveryConfig, kubeClient clientset.Interface,
    vcClient vcclientset.Interface) (Discoverer, error) {
    constructor, exists := discovererRegistry[cfg.Source]
    if !exists {
        return nil, fmt.Errorf("unsupported discoverer type: %s", cfg.Source)
    }
    return constructor(cfg, kubeClient, vcClient), nil
}

type DiscoveryConfig struct {
    Source      string                 `json:"source" yaml:"source"`
    Enabled     bool                   `json:"enabled" yaml:"enabled"`
    Interval    time.Duration          `json:"interval" yaml:"interval"`
    Credentials *Credentials           `json:"credentials" yaml:"credentials"`
    Config      map[string]interface{} `json:"config" yaml:"config"`
}

package custom

import "volcano.sh/volcano/pkg/controllers/hypernode/api"

func init() {
    api.RegisterDiscoverer("custom", NewCustomDiscoverer)
}

type customDiscoverer struct { /* ... */ }

func NewCustomDiscoverer(cfg api.DiscoveryConfig, kubeClient clientset.Interface,
    vcClient vcclientset.Interface) api.Discoverer {
    return &customDiscoverer{ /* ... */ }
}

func (c *customDiscoverer) Start() (chan []*topologyv1alpha1.HyperNode, error) { /* ... */ }
func (c *customDiscoverer) Stop() error { /* ... */ }
func (c *customDiscoverer) Name() string { return "custom" }
func (c *customDiscoverer) ResultSynced() { /* ... */ }

func (hn *hyperNodeController) setConfigMapNamespaceAndName() {
    namespace := os.Getenv("KUBE_POD_NAMESPACE")     // 默认: "volcano-system"
    releaseName := os.Getenv("HELM_RELEASE_NAME")     // 默认: "volcano"
    hn.configMapNamespace = namespace
    hn.configMapName = releaseName + "-controller-configmap"
}

func (hn *hyperNodeController) setupConfigMapInformer() {
    filteredInformer := coreinformers.NewFilteredConfigMapInformer(
        hn.kubeClient,
        hn.configMapNamespace,
        0,
        cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
        func(options *metav1.ListOptions) {
            options.FieldSelector = fmt.Sprintf("metadata.name=%s", hn.configMapName)
        },
    )
    // ...
}

type Loader interface {
    LoadConfig() (*api.NetworkTopologyConfig, error)
}

# volcano-controller-configmap 中的 volcano-controller.conf
networkTopologyDiscovery:
  - source: ufm
    enabled: true
    interval: 10m
    credentials:
      secretRef:
        name: ufm-credentials
        namespace: volcano-system
    config:
      endpoint: https://ufm-server:8080
      insecureSkipVerify: true
  - source: label
    enabled: true
    config:
      networkTopologyTypes:
        topologyA2:
          - nodeLabel: "volcano.sh/tor"
          - nodeLabel: "kubernetes.io/hostname"
        topologyA3:
          - nodeLabel: "volcano.sh/hypercluster"
          - nodeLabel: "volcano.sh/hypernode"
          - nodeLabel: "kubernetes.io/hostname"

func (hn *hyperNodeController) addHyperNode(obj interface{}) {
    hyperNode, ok := obj.(*topologyv1alpha1.HyperNode)
    if !ok { return }
    hn.enqueueHyperNode(hyperNode)
}

func (hn *hyperNodeController) updateHyperNode(oldObj, newObj interface{}) {
    hyperNode, ok := newObj.(*topologyv1alpha1.HyperNode)
    if !ok { return }
    hn.enqueueHyperNode(hyperNode)
}

func (hn *hyperNodeController) enqueueHyperNode(hyperNode *topologyv1alpha1.HyperNode) {
    key, err := cache.MetaNamespaceKeyFunc(hyperNode)
    if err != nil { return }
    hn.hyperNodeQueue.Add(key)
}

func (hn *hyperNodeController) processHyperNodeQueue() {
    for {
        key, shutdown := hn.hyperNodeQueue.Get()
        if shutdown { return }
        func() {
            defer hn.hyperNodeQueue.Done(key)
            err := hn.syncHyperNodeStatus(key)
            if err != nil {
                hn.hyperNodeQueue.AddRateLimited(key)
                return
            }
            hn.hyperNodeQueue.Forget(key)
        }()
    }
}

func (hn *hyperNodeController) actualNodeCnt(hyperNode *topologyv1alpha1.HyperNode) int {
    nodes, err := hn.nodeLister.List(labels.Everything())
    if err != nil { return 0 }
    members := sets.New[string]()
    for _, member := range hyperNode.Spec.Members {
        members.Insert(api.GetMembers(member.Selector, nodes).UnsortedList()...)
    }
    return len(members)
}

func (hn *hyperNodeController) watchDiscoveryResults() {
    resultCh := hn.discoveryManager.ResultChannel()
    for result := range resultCh {
        if result.HyperNodes != nil {
            hn.reconcileTopology(result.Source, result.HyperNodes)
            hn.discoveryManager.ResultSynced(result.Source)
        }
    }
}

const NetworkTopologySourceLabelKey = "volcano.sh/network-topology-source"

// +genclient:nonNamespaced
// +kubebuilder:resource:path=hypernodes,shortName=hn,scope=Cluster
// +kubebuilder:subresource:status
type HyperNode struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`
    Spec              HyperNodeSpec   `json:"spec"`
    Status            HyperNodeStatus `json:"status,omitempty"`
}

type HyperNodeSpec struct {
    Tier     int          `json:"tier,omitempty"`
    TierName string       `json:"tierName,omitempty"`
    Members  []MemberSpec `json:"members,omitempty"`
}

type MemberSpec struct {
    Type     MemberType     `json:"type,omitempty"`
    Selector MemberSelector `json:"selector,omitempty"`
}

type MemberSelector struct {
    ExactMatch *ExactMatch            `json:"exactMatch,omitempty"`
    RegexMatch *RegexMatch            `json:"regexMatch,omitempty"`
    LabelMatch *metav1.LabelSelector  `json:"labelMatch,omitempty"`
}

type HyperNodeStatus struct {
    Conditions []metav1.Condition `json:"conditions,omitempty"`
    NodeCount  int64              `json:"nodeCount,omitempty"`
}

func CreateHyperNode(vcClient vcclientset.Interface, node *topologyv1alpha1.HyperNode) error
func UpdateHyperNode(vcClient vcclientset.Interface, lister v1alpha1.HyperNodeLister,
    updated *topologyv1alpha1.HyperNode) error
func DeleteHyperNode(vcClient vcclientset.Interface, name string) error

func BuildHyperNode(name string, tier int, members []topologyv1alpha1.MemberSpec,
    labels map[string]string) *topologyv1alpha1.HyperNode

func BuildHyperNodeWithTierName(name string, tier int, tierName string,
    members []topologyv1alpha1.MemberSpec, labels map[string]string) *topologyv1alpha1.HyperNode

func BuildMembers(names []string, memberType topologyv1alpha1.MemberType) []topologyv1alpha1.MemberSpec {
    members := make([]topologyv1alpha1.MemberSpec, 0, len(names))
    sort.Strings(names)
    for _, name := range names {
        members = append(members, topologyv1alpha1.MemberSpec{
            Type:     memberType,
            Selector: topologyv1alpha1.MemberSelector{
                ExactMatch: &topologyv1alpha1.ExactMatch{Name: name},
            },
        })
    }
    return members
}

hn.hyperNodeQueue = workqueue.NewTypedRateLimitingQueue(
    workqueue.DefaultTypedControllerRateLimiter[string]())
hn.configMapQueue = workqueue.NewTypedRateLimitingQueue(
    workqueue.DefaultTypedControllerRateLimiter[string]())

for informerType, ok := range hn.informerFactory.WaitForCacheSync(stopCh) {
    if !ok {
        klog.ErrorS(nil, "Failed to sync informer cache: %v", informerType)
        return
    }
}

func (m *manager) Start() error {
    cfg, err := m.configLoader.LoadConfig()
    if err != nil {
        klog.ErrorS(err, "Failed to load config")
        m.config = &api.NetworkTopologyConfig{}
        // 不返回错误，继续运行
    }
    // ...
}

func CreateHyperNode(vcClient vcclientset.Interface, node *topologyv1alpha1.HyperNode) error {
    return retry.RetryOnConflict(retry.DefaultRetry, func() error {
        _, err := vcClient.TopologyV1alpha1().HyperNodes().Create(
            context.Background(), node, metav1.CreateOptions{})
        return err
    })
}

func (l *labelDiscoverer) work() {
    for {
        key, shutdown := l.queue.Get()
        if shutdown { return }
        func() {
            defer l.queue.Done(key)
            if err := l.discovery(); err != nil {
                l.queue.AddRateLimited(key)
                return
            }
            l.queue.Forget(key)
        }()
        <-l.completedCh  // 等待上一次结果被处理完成
    }
}