Skip to content

Commit

Permalink
feat: support ha mode for local registry (#1901)
Browse files Browse the repository at this point in the history 
* feat: support recover master0

* feat: support ha mode for local registry

update reviews

update reviews
  • Loading branch information
@kakaZhou719
kakaZhou719 authored Dec 15, 2022
1 parent 896619e commit ce6e62a
Show file tree
Hide file tree
Showing 32 changed files with 1,064 additions and 603 deletions.
89 changes: 26 additions & 63 deletions cmd/sealer/cmd/cluster/apply.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,30 +16,24 @@ package cluster

import (
"fmt"
"net"
"os"
"path/filepath"

"github.com/pkg/errors"
"github.com/sealerio/sealer/cmd/sealer/cmd/types"
"github.com/sealerio/sealer/cmd/sealer/cmd/utils"
"github.com/sealerio/sealer/common"
"github.com/sealerio/sealer/pkg/client/k8s"
"github.com/sealerio/sealer/pkg/clusterfile"
v12 "github.com/sealerio/sealer/pkg/define/image/v1"
"github.com/sealerio/sealer/pkg/define/options"
"github.com/sealerio/sealer/pkg/imageengine"
"github.com/sealerio/sealer/pkg/infradriver"
v2 "github.com/sealerio/sealer/types/api/v2"
"github.com/sealerio/sealer/utils/strings"
"github.com/sirupsen/logrus"
"github.com/spf13/cobra"
corev1 "k8s.io/api/core/v1"
)

var applyClusterFile string

var applyMode string

const MasterRoleLabel = "node-role.kubernetes.io/master"
var applyFlags *types.ApplyFlags

var longApplyCmdDescription = `apply command is used to apply a Kubernetes cluster via specified Clusterfile.
If the Clusterfile is applied first time, Kubernetes cluster will be created. Otherwise, sealer
Expand All @@ -58,9 +52,11 @@ func NewApplyCmd() *cobra.Command {
Args: cobra.NoArgs,
RunE: func(cmd *cobra.Command, args []string) error {
var (
cf clusterfile.Interface
clusterFileData []byte
err error
cf clusterfile.Interface
clusterFileData []byte
err error
applyClusterFile = applyFlags.ClusterFile
applyMode = applyFlags.ApplyMode
)
logrus.Warn("sealer apply command will be deprecated in the future, please use sealer run instead.")

Expand Down Expand Up @@ -92,24 +88,33 @@ func NewApplyCmd() *cobra.Command {
return err
}

if err = imageEngine.Pull(&options.PullOptions{
Quiet: false,
PullPolicy: "missing",
Image: imageName,
Platform: "local",
}); err != nil {
return err
}

extension, err := imageEngine.GetSealerImageExtension(&options.GetImageAnnoOptions{ImageNameOrID: imageName})
if err != nil {
return fmt.Errorf("failed to get cluster image extension: %s", err)
}

if extension.Type == v12.AppInstaller {
logrus.Infof("start to install application: %s", imageName)
return installApplication(imageName, []string{}, extension, infraDriver, imageEngine)
return installApplication(imageName, []string{}, extension, infraDriver, imageEngine, applyMode)
}

client := getClusterClient()
client := utils.GetClusterClient()
if client == nil {
// no k8s client means to init a new cluster.
logrus.Infof("start to create new cluster with image: %s", imageName)
return createNewCluster(imageName, infraDriver, imageEngine, cf)
return createNewCluster(infraDriver, imageEngine, cf, applyMode)
}

currentCluster, err := GetCurrentCluster(client)
currentCluster, err := utils.GetCurrentCluster(client)
if err != nil {
return errors.Wrap(err, "failed to get current cluster")
}
Expand All @@ -127,52 +132,10 @@ func NewApplyCmd() *cobra.Command {
return scaleUpCluster(imageName, mj, nj, infraDriver, imageEngine, cf)
},
}
applyCmd.Flags().BoolVar(&ForceDelete, "force", false, "force to delete the specified cluster if set true")
applyCmd.Flags().StringVarP(&applyClusterFile, "Clusterfile", "f", "", "Clusterfile path to apply a Kubernetes cluster")
applyCmd.Flags().StringVarP(&applyMode, "applyMode", "m", common.ApplyModeApply, "load images to the specified registry in advance")
return applyCmd
}

func GetCurrentCluster(client *k8s.Client) (*v2.Cluster, error) {
nodes, err := client.ListNodes()
if err != nil {
return nil, err
}

cluster := &v2.Cluster{}
var masterIPList []net.IP
var nodeIPList []net.IP

for _, node := range nodes.Items {
addr := getNodeAddress(node)
if addr == nil {
continue
}
if _, ok := node.Labels[MasterRoleLabel]; ok {
masterIPList = append(masterIPList, addr)
continue
}
nodeIPList = append(nodeIPList, addr)
}
cluster.Spec.Hosts = []v2.Host{{IPS: masterIPList, Roles: []string{common.MASTER}}, {IPS: nodeIPList, Roles: []string{common.NODE}}}

return cluster, nil
}

func getNodeAddress(node corev1.Node) net.IP {
if len(node.Status.Addresses) < 1 {
return nil
}
return net.ParseIP(node.Status.Addresses[0].Address)
}

func getClusterClient() *k8s.Client {
client, err := k8s.NewK8sClient()
if client != nil {
return client
}
if err != nil {
logrus.Warnf("try to new k8s client via default kubeconfig, maybe this is a new cluster that needs to be created: %v", err)
}
return nil
applyFlags = &types.ApplyFlags{}
applyCmd.Flags().BoolVar(&applyFlags.ForceDelete, "force", false, "force to delete the specified cluster if set true")
applyCmd.Flags().StringVarP(&applyFlags.ClusterFile, "Clusterfile", "f", "", "Clusterfile path to apply a Kubernetes cluster")
applyCmd.Flags().StringVarP(&applyFlags.ApplyMode, "applyMode", "m", common.ApplyModeApply, "load images to the specified registry in advance")
return applyCmd
}
53 changes: 22 additions & 31 deletions cmd/sealer/cmd/cluster/run-app.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@ package cluster

import (
"fmt"
"path/filepath"

"github.com/sealerio/sealer/cmd/sealer/cmd/types"
"github.com/sealerio/sealer/common"
Expand All @@ -27,7 +26,6 @@ import (
"github.com/sealerio/sealer/pkg/imagedistributor"
"github.com/sealerio/sealer/pkg/imageengine"
"github.com/sealerio/sealer/pkg/infradriver"
"github.com/sealerio/sealer/pkg/registry"
"github.com/sirupsen/logrus"
"github.com/spf13/cobra"
)
Expand All @@ -37,16 +35,17 @@ var longNewRunAPPCmdDescription = `sealer run-app localhost/nginx:v1`
var exampleForRunAppCmd = ``

func NewRunAPPCmd() *cobra.Command {
runCmd := &cobra.Command{
runAppCmd := &cobra.Command{
Use: "run-app",
Short: "start to run an application cluster image",
Long: longNewRunAPPCmdDescription,
Example: exampleForRunAppCmd,
Args: cobra.ExactArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
var (
cf clusterfile.Interface
err error
cf clusterfile.Interface
err error
applyMode = appFlags.ApplyMode
)

cf, err = clusterfile.NewClusterFile(nil)
Expand All @@ -65,6 +64,15 @@ func NewRunAPPCmd() *cobra.Command {
return err
}

if err = imageEngine.Pull(&options.PullOptions{
Quiet: false,
PullPolicy: "missing",
Image: args[0],
Platform: "local",
}); err != nil {
return err
}

extension, err := imageEngine.GetSealerImageExtension(&options.GetImageAnnoOptions{ImageNameOrID: args[0]})
if err != nil {
return fmt.Errorf("failed to get cluster image extension: %s", err)
Expand All @@ -74,18 +82,20 @@ func NewRunAPPCmd() *cobra.Command {
return fmt.Errorf("exit install process, wrong cluster image type: %s", extension.Type)
}

return installApplication(args[0], appFlags.LaunchCmds, extension, infraDriver, imageEngine)
return installApplication(args[0], appFlags.LaunchCmds, extension, infraDriver, imageEngine, applyMode)
},
}

appFlags = &types.APPFlags{}
runCmd.Flags().StringSliceVar(&appFlags.LaunchCmds, "cmds", []string{}, "override default LaunchCmds of clusterimage")
runAppCmd.Flags().StringSliceVar(&appFlags.LaunchCmds, "cmds", []string{}, "override default LaunchCmds of clusterimage")
//runCmd.Flags().StringSliceVar(&appFlags.LaunchArgs, "args", []string{}, "override default LaunchArgs of clusterimage")
return runCmd
runAppCmd.Flags().StringVarP(&appFlags.ApplyMode, "applyMode", "m", common.ApplyModeApply, "load images to the specified registry in advance")

return runAppCmd
}

func installApplication(appImageName string, launchCmds []string, extension v12.ImageExtension,
infraDriver infradriver.InfraDriver, imageEngine imageengine.Interface) error {
infraDriver infradriver.InfraDriver, imageEngine imageengine.Interface, mode string) error {
clusterHosts := infraDriver.GetHostIPList()

clusterHostsPlatform, err := infraDriver.GetHostsPlatform(clusterHosts)
Expand Down Expand Up @@ -115,30 +125,11 @@ func installApplication(appImageName string, launchCmds []string, extension v12.
return err
}

if applyMode == common.ApplyModeLoadImage {
logrus.Infof("start to apply with mode(%s)", applyMode)

if err = distributor.DistributeRegistry(infraDriver.GetHostIPList()[0], infraDriver.GetClusterRootfsPath()); err != nil {
return err
}
logrus.Infof("load image success")
return nil
}

//todo grab this config from cluster file, that's because it belongs to cluster level information
var registryConfig registry.RegConfig
var config = registry.Registry{
Domain: registry.DefaultDomain,
Port: registry.DefaultPort,
}

registryConfig.LocalRegistry = &registry.LocalRegistry{
DataDir: filepath.Join(infraDriver.GetClusterRootfsPath(), "registry"),
DeployHost: infraDriver.GetHostIPListByRole(common.MASTER)[0],
Registry: config,
if mode == common.ApplyModeLoadImage {
return loadToRegistry(infraDriver, distributor)
}

installer := clusterruntime.NewAppInstaller(infraDriver, distributor, extension, registryConfig)
installer := clusterruntime.NewAppInstaller(infraDriver, distributor, extension)
err = installer.Install(infraDriver.GetHostIPListByRole(common.MASTER)[0], launchCmds)
if err != nil {
return err
Expand Down
62 changes: 43 additions & 19 deletions cmd/sealer/cmd/cluster/run.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ package cluster

import (
"fmt"
"net"
"os"
"path/filepath"

Expand All @@ -36,7 +37,6 @@ import (
)

var runFlags *types.Flags
var clusterFile string

var longNewRunCmdDescription = `sealer run registry.cn-qingdao.aliyuncs.com/sealer-io/kubernetes:v1.19.8 --masters [arg] --nodes [arg]`

Expand Down Expand Up @@ -70,7 +70,10 @@ func NewRunCmd() *cobra.Command {
cf clusterfile.Interface
clusterFileData []byte
err error
clusterFile = runFlags.ClusterFile
applyMode = runFlags.Mode
)

if runFlags.Masters == "" && clusterFile == "" {
return fmt.Errorf("you must input master ip Or use Clusterfile")
}
Expand Down Expand Up @@ -121,7 +124,7 @@ func NewRunCmd() *cobra.Command {
return err
}

return createNewCluster(cluster.Spec.Image, infraDriver, imageEngine, cf)
return createNewCluster(infraDriver, imageEngine, cf, applyMode)
},
}
runFlags = &types.Flags{}
Expand All @@ -136,7 +139,8 @@ func NewRunCmd() *cobra.Command {
runCmd.Flags().StringVar(&runFlags.PkPassword, "pk-passwd", "", "set baremetal server private key password")
runCmd.Flags().StringSliceVar(&runFlags.CMDArgs, "cmd-args", []string{}, "set args for image cmd instruction")
runCmd.Flags().StringSliceVarP(&runFlags.CustomEnv, "env", "e", []string{}, "set custom environment variables")
runCmd.Flags().StringVarP(&clusterFile, "Clusterfile", "f", "", "Clusterfile path to run a Kubernetes cluster")
runCmd.Flags().StringVarP(&runFlags.ClusterFile, "Clusterfile", "f", "", "Clusterfile path to run a Kubernetes cluster")
runCmd.Flags().StringVar(&runFlags.Mode, "mode", common.ApplyModeApply, "load images to the specified registry in advance")

//err := runCmd.RegisterFlagCompletionFunc("provider", func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
// return strings.ContainPartial([]string{common.BAREMETAL, common.AliCloud, common.CONTAINER}, toComplete), cobra.ShellCompDirectiveNoFileComp
Expand All @@ -148,10 +152,10 @@ func NewRunCmd() *cobra.Command {
return runCmd
}

func createNewCluster(clusterImageName string, infraDriver infradriver.InfraDriver, imageEngine imageengine.Interface, cf clusterfile.Interface) error {
func createNewCluster(infraDriver infradriver.InfraDriver, imageEngine imageengine.Interface, cf clusterfile.Interface, mode string) error {
var (
clusterHosts = infraDriver.GetHostIPList()
clusterLaunchCmds = infraDriver.GetClusterLaunchCmds()
clusterHosts = infraDriver.GetHostIPList()
clusterImageName = infraDriver.GetClusterImageName()
)

clusterHostsPlatform, err := infraDriver.GetHostsPlatform(clusterHosts)
Expand Down Expand Up @@ -181,15 +185,10 @@ func createNewCluster(clusterImageName string, infraDriver infradriver.InfraDriv
return err
}

if applyMode == common.ApplyModeLoadImage {
logrus.Infof("start to apply with mode(%s)", applyMode)

if err = distributor.DistributeRegistry(infraDriver.GetHostIPList()[0], infraDriver.GetClusterRootfsPath()); err != nil {
return err
}
logrus.Infof("load image success")
return nil
if mode == common.ApplyModeLoadImage {
return loadToRegistry(infraDriver, distributor)
}

plugins, err := loadPluginsFromImage(imageMountInfo)
if err != nil {
return err
Expand All @@ -200,11 +199,9 @@ func createNewCluster(clusterImageName string, infraDriver infradriver.InfraDriv
}

runtimeConfig := &clusterruntime.RuntimeConfig{
Distributor: distributor,
ImageEngine: imageEngine,
Plugins: plugins,
ClusterLaunchCmds: clusterLaunchCmds,
ClusterImageImage: clusterImageName,
Distributor: distributor,
ImageEngine: imageEngine,
Plugins: plugins,
}

if cf.GetKubeadmConfig() != nil {
Expand Down Expand Up @@ -248,3 +245,30 @@ func loadPluginsFromImage(imageMountInfo []imagedistributor.ClusterImageMountInf

return plugins, nil
}

// loadToRegistry just load container image to local registry
func loadToRegistry(infraDriver infradriver.InfraDriver, distributor imagedistributor.Distributor) error {
regConfig := infraDriver.GetClusterRegistryConfig()
// todo only support load image to local registry at present
if regConfig.LocalRegistry == nil {
return nil
}

deployHosts := infraDriver.GetHostIPListByRole(common.MASTER)
if len(deployHosts) < 1 {
return fmt.Errorf("local registry host can not be nil")
}
master0 := deployHosts[0]

logrus.Infof("start to apply with mode(%s)", common.ApplyModeLoadImage)
if !regConfig.LocalRegistry.HaMode {
deployHosts = []net.IP{master0}
}

if err := distributor.DistributeRegistry(deployHosts, filepath.Join(infraDriver.GetClusterRootfsPath(), "registry")); err != nil {
return err
}

logrus.Infof("load image success")
return nil
}
Loading

0 comments on commit ce6e62a

Please sign in to comment.