代码拉取完成,页面将自动刷新
package clusterdeploy
import (
"bytes"
"context"
"errors"
"fmt"
"reflect"
"time"
"github.com/rancher/norman/types"
"github.com/rancher/rancher/pkg/clustermanager"
"github.com/rancher/rancher/pkg/image"
"github.com/rancher/rancher/pkg/kubectl"
"github.com/rancher/rancher/pkg/settings"
"github.com/rancher/rancher/pkg/systemaccount"
"github.com/rancher/rancher/pkg/systemtemplate"
"github.com/rancher/types/apis/management.cattle.io/v3"
"github.com/rancher/types/config"
"github.com/rancher/types/user"
"github.com/sirupsen/logrus"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/runtime"
clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
)
func Register(ctx context.Context, management *config.ManagementContext, clusterManager *clustermanager.Manager) {
c := &clusterDeploy{
systemAccountManager: systemaccount.NewManager(management),
userManager: management.UserManager,
clusters: management.Management.Clusters(""),
nodeLister: management.Management.Nodes("").Controller().Lister(),
clusterManager: clusterManager,
}
management.Management.Clusters("").AddHandler(ctx, "cluster-deploy", c.sync)
}
type clusterDeploy struct {
systemAccountManager *systemaccount.Manager
userManager user.Manager
clusters v3.ClusterInterface
clusterManager *clustermanager.Manager
nodeLister v3.NodeLister
}
func (cd *clusterDeploy) sync(key string, cluster *v3.Cluster) (runtime.Object, error) {
var (
err, updateErr error
)
if key == "" || cluster == nil {
return nil, nil
}
original := cluster
cluster = original.DeepCopy()
if cluster.Status.Driver == v3.ClusterDriverRKE {
if cluster.Spec.LocalClusterAuthEndpoint.Enabled {
cluster.Spec.RancherKubernetesEngineConfig.Authentication.Strategy = "x509|webhook"
} else {
cluster.Spec.RancherKubernetesEngineConfig.Authentication.Strategy = "x509"
}
}
err = cd.doSync(cluster)
if cluster != nil && !reflect.DeepEqual(cluster, original) {
_, updateErr = cd.clusters.Update(cluster)
}
if err != nil {
return nil, err
}
return nil, updateErr
}
func (cd *clusterDeploy) doSync(cluster *v3.Cluster) error {
if !v3.ClusterConditionProvisioned.IsTrue(cluster) {
return nil
}
nodes, err := cd.nodeLister.List(cluster.Name, labels.Everything())
if err != nil {
return err
}
if len(nodes) == 0 {
return nil
}
_, err = v3.ClusterConditionSystemAccountCreated.DoUntilTrue(cluster, func() (runtime.Object, error) {
return cluster, cd.systemAccountManager.CreateSystemAccount(cluster)
})
if err != nil {
return err
}
err = cd.deployAgent(cluster)
if err != nil {
return err
}
return cd.setNetworkPolicyAnn(cluster)
}
func (cd *clusterDeploy) deployAgent(cluster *v3.Cluster) error {
desiredAgent := cluster.Spec.DesiredAgentImage
if desiredAgent == "" || desiredAgent == "fixed" {
desiredAgent = image.Resolve(settings.AgentImage.Get())
}
var desiredAuth string
if cluster.Spec.LocalClusterAuthEndpoint.Enabled {
desiredAuth = cluster.Spec.DesiredAuthImage
if desiredAuth == "" || desiredAuth == "fixed" {
desiredAuth = image.Resolve(settings.AuthImage.Get())
}
}
if cluster.Status.AgentImage == desiredAgent && cluster.Status.AuthImage == desiredAuth {
return nil
}
kubeConfig, err := cd.getKubeConfig(cluster)
if err != nil {
return err
}
_, err = v3.ClusterConditionAgentDeployed.Do(cluster, func() (runtime.Object, error) {
yaml, err := cd.getYAML(cluster, desiredAgent, desiredAuth)
if err != nil {
return cluster, err
}
var output []byte
for i := 0; i < 3; i++ {
// This will fail almost always the first time because when we create the namespace in the file
// it won't have privileges. Just stupidly try 3 times
output, err = kubectl.Apply(yaml, kubeConfig)
if err == nil {
break
}
time.Sleep(2 * time.Second)
}
if err != nil {
return cluster, types.NewErrors(err, errors.New(string(output)))
}
v3.ClusterConditionAgentDeployed.Message(cluster, string(output))
if !cluster.Spec.LocalClusterAuthEndpoint.Enabled && cluster.Status.AppliedSpec.LocalClusterAuthEndpoint.Enabled && cluster.Status.AuthImage != "" {
output, err = kubectl.Delete([]byte(systemtemplate.AuthDaemonSet), kubeConfig)
}
if err != nil {
return cluster, types.NewErrors(err, errors.New(string(output)))
}
v3.ClusterConditionAgentDeployed.Message(cluster, string(output))
return cluster, nil
})
if err != nil {
return err
}
if err == nil {
cluster.Status.AgentImage = desiredAgent
if cluster.Spec.DesiredAgentImage == "fixed" {
cluster.Spec.DesiredAgentImage = desiredAgent
}
cluster.Status.AuthImage = desiredAuth
if cluster.Spec.DesiredAuthImage == "fixed" {
cluster.Spec.DesiredAuthImage = desiredAuth
}
}
return err
}
func (cd *clusterDeploy) setNetworkPolicyAnn(cluster *v3.Cluster) error {
if cluster.Spec.EnableNetworkPolicy != nil {
return nil
}
// set current state for upgraded canal clusters
if cluster.Spec.RancherKubernetesEngineConfig != nil &&
cluster.Spec.RancherKubernetesEngineConfig.Network.Plugin == "canal" {
enableNetworkPolicy := true
cluster.Spec.EnableNetworkPolicy = &enableNetworkPolicy
cluster.Annotations["networking.management.cattle.io/enable-network-policy"] = "true"
}
return nil
}
func (cd *clusterDeploy) getKubeConfig(cluster *v3.Cluster) (*clientcmdapi.Config, error) {
user, err := cd.systemAccountManager.GetSystemUser(cluster)
if err != nil {
return nil, err
}
token, err := cd.userManager.EnsureToken("agent-"+user.Name, "token for agent deployment", user.Name)
if err != nil {
return nil, err
}
return cd.clusterManager.KubeConfig(cluster.Name, token), nil
}
func (cd *clusterDeploy) getYAML(cluster *v3.Cluster, agentImage, authImage string) ([]byte, error) {
logrus.Debug("Desired agent image:", agentImage)
logrus.Debug("Desired auth image:", authImage)
token, err := cd.systemAccountManager.GetOrCreateSystemClusterToken(cluster.Name)
if err != nil {
return nil, err
}
url := settings.ServerURL.Get()
if url == "" {
return nil, fmt.Errorf("waiting for server-url setting to be set")
}
buf := &bytes.Buffer{}
err = systemtemplate.SystemTemplate(buf, agentImage, authImage, token, url)
return buf.Bytes(), err
}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。