diff --git a/Documentation/cluster-crd.md b/Documentation/cluster-crd.md index 6162e9d9b..21db20378 100644 --- a/Documentation/cluster-crd.md +++ b/Documentation/cluster-crd.md @@ -23,6 +23,7 @@ Settings can be specified at the global level to apply to the cluster as a whole - `versionTag`: The version (tag) of the `rook/rook` container that will be deployed. Upgrades are not yet supported if this setting is updated for an existing cluster, but upgrades will be coming. - `dataDirHostPath`: The host path where config and data should be stored for each of the services. If the directory does not exist, it will be created. Because this directory persists on the host, it will remain after pods are deleted. Therefore, for test scenarios, the path must be deleted if you are going to delete a cluster and start a new cluster on the same hosts. More details can be found in the Kubernetes [host path docs](https://kubernetes.io/docs/concepts/storage/volumes/#hostpath). If this value is empty, each pod will get an ephemeral directory to store their config files that is tied to the lifetime of the pod running on that node. More details can be found in the Kubernetes [empty dir docs](https://kubernetes.io/docs/concepts/storage/volumes/#emptydir). +- `hostNetwork`: uses host network instead of using the SDN below the containers - `placement`: [placement configuration settings](#placement-configuration-settings) - `storage`: Storage selection and configuration that will be used across the cluster. Note that these settings can be overridden for specific nodes. - `useAllNodes`: `true` or `false`, indicating if all nodes in the cluster should be used for storage according to the cluster level storage selection and configuration values. diff --git a/cluster/examples/kubernetes/rook-cluster.yaml b/cluster/examples/kubernetes/rook-cluster.yaml index 59e701d9f..efbdd5684 100644 --- a/cluster/examples/kubernetes/rook-cluster.yaml +++ b/cluster/examples/kubernetes/rook-cluster.yaml @@ -11,6 +11,8 @@ metadata: spec: versionTag: master dataDirHostPath: + # toggle to use hostNetwork + hostNetwork: false # To control where various services will be scheduled by kubernetes, use the placement configuration sections below. # The example under 'all' would have all services scheduled on kubernetes nodes labeled with 'role=storage' and # tolerate taints with a key of 'storage-node'. diff --git a/cluster/examples/kubernetes/rook-tools.yaml b/cluster/examples/kubernetes/rook-tools.yaml index 02eab81bc..c8c06d4f8 100644 --- a/cluster/examples/kubernetes/rook-tools.yaml +++ b/cluster/examples/kubernetes/rook-tools.yaml @@ -4,6 +4,7 @@ metadata: name: rook-tools namespace: rook spec: + hostNetwork: false containers: - name: rook-tools image: rook/toolbox:master @@ -33,4 +34,4 @@ spec: path: /sys/bus - name: libmodules hostPath: - path: /lib/modules \ No newline at end of file + path: /lib/modules diff --git a/cmd/rook/api.go b/cmd/rook/api.go index 8973c1976..d47a77cad 100644 --- a/cmd/rook/api.go +++ b/cmd/rook/api.go @@ -31,10 +31,11 @@ var ( Use: "api", Short: "Runs the Rook API service", } - apiPort int - repoPrefix string - namespace string - versionTag string + apiPort int + repoPrefix string + namespace string + versionTag string + hostNetwork bool ) func init() { @@ -42,6 +43,7 @@ func init() { apiCmd.Flags().StringVar(&repoPrefix, "repo-prefix", "rook", "the repo from which to pull images") apiCmd.Flags().StringVar(&versionTag, "version-tag", "latest", "version of the rook container to launch") apiCmd.Flags().StringVar(&namespace, "namespace", "", "the namespace in which the api service is running") + apiCmd.Flags().BoolVar(&hostNetwork, "hostnetwork", false, "if the hostnetwork should be used") addCephFlags(apiCmd) flags.SetFlagsFromEnv(apiCmd.Flags(), RookEnvVarPrefix) @@ -74,7 +76,7 @@ func startAPI(cmd *cobra.Command, args []string) error { apiCfg := &api.Config{ Port: apiPort, ClusterInfo: &clusterInfo, - ClusterHandler: apik8s.New(context, &clusterInfo, namespace, versionTag), + ClusterHandler: apik8s.New(context, &clusterInfo, namespace, versionTag, hostNetwork), } err = api.Run(context, apiCfg) diff --git a/cmd/rook/mon.go b/cmd/rook/mon.go index 336ae7535..10043b0d7 100644 --- a/cmd/rook/mon.go +++ b/cmd/rook/mon.go @@ -32,7 +32,7 @@ var monCmd = &cobra.Command{ var ( monName string - monPort int + monPort int32 ) func addCephFlags(command *cobra.Command) { @@ -49,7 +49,7 @@ func addCephFlags(command *cobra.Command) { func init() { monCmd.Flags().StringVar(&monName, "name", "", "name of the monitor") - monCmd.Flags().IntVar(&monPort, "port", 0, "port of the monitor") + monCmd.Flags().Int32Var(&monPort, "port", 0, "port of the monitor") addCephFlags(monCmd) flags.SetFlagsFromEnv(monCmd.Flags(), RookEnvVarPrefix) @@ -73,9 +73,13 @@ func startMon(cmd *cobra.Command, args []string) error { // at first start the local monitor needs to be added to the list of mons clusterInfo.Monitors = mon.ParseMonEndpoints(cfg.monEndpoints) - clusterInfo.Monitors[monName] = mon.ToCephMon(monName, cfg.networkInfo.PublicAddrIPv4) + clusterInfo.Monitors[monName] = mon.ToCephMon(monName, cfg.networkInfo.PublicAddrIPv4, monPort) - monCfg := &mon.Config{Name: monName, Cluster: &clusterInfo} + monCfg := &mon.Config{ + Name: monName, + Cluster: &clusterInfo, + Port: monPort, + } err := mon.Run(createContext(), monCfg) if err != nil { fmt.Fprintln(os.Stderr, err) diff --git a/pkg/api/k8s/cluster_handler.go b/pkg/api/k8s/cluster_handler.go index 1aa82cb1b..7dd1573f3 100644 --- a/pkg/api/k8s/cluster_handler.go +++ b/pkg/api/k8s/cluster_handler.go @@ -35,10 +35,11 @@ type clusterHandler struct { clusterInfo *mon.ClusterInfo namespace string versionTag string + hostNetwork bool } -func New(context *clusterd.Context, clusterInfo *mon.ClusterInfo, namespace, versionTag string) *clusterHandler { - return &clusterHandler{context: context, clusterInfo: clusterInfo, namespace: namespace, versionTag: versionTag} +func New(context *clusterd.Context, clusterInfo *mon.ClusterInfo, namespace, versionTag string, hostNetwork bool) *clusterHandler { + return &clusterHandler{context: context, clusterInfo: clusterInfo, namespace: namespace, versionTag: versionTag, hostNetwork: hostNetwork} } func (s *clusterHandler) GetClusterInfo() (*mon.ClusterInfo, error) { @@ -66,7 +67,7 @@ func (s *clusterHandler) EnableObjectStore(config model.ObjectStore) error { // Passing an empty Placement{} as the api doesn't know about placement // information. This should be resolved with the transition to CRD (TPR). - r := k8srgw.New(s.context, config, s.namespace, s.versionTag, k8sutil.Placement{}) + r := k8srgw.New(s.context, config, s.namespace, s.versionTag, k8sutil.Placement{}, s.hostNetwork) err := r.Start() if err != nil { return fmt.Errorf("failed to start rgw. %+v", err) @@ -98,7 +99,7 @@ func (s *clusterHandler) StartFileSystem(fs *model.FilesystemRequest) error { logger.Infof("Starting the MDS") // Passing an empty Placement{} as the api doesn't know about placement // information. This should be resolved with the transition to CRD (TPR). - c := k8smds.New(s.context, s.namespace, s.versionTag, k8sutil.Placement{}) + c := k8smds.New(s.context, s.namespace, s.versionTag, k8sutil.Placement{}, s.hostNetwork) return c.Start() } diff --git a/pkg/ceph/client/image.go b/pkg/ceph/client/image.go index 3f6d8b207..3e5803ec2 100644 --- a/pkg/ceph/client/image.go +++ b/pkg/ceph/client/image.go @@ -21,8 +21,9 @@ import ( "strconv" - "github.com/rook/rook/pkg/clusterd" "regexp" + + "github.com/rook/rook/pkg/clusterd" ) const ( @@ -44,17 +45,17 @@ func ListImages(context *clusterd.Context, clusterName, poolName string) ([]Ceph //The regex expression captures the json result at the end buf //When logLevel is DEBUG buf contains log statements of librados (see tests for examples) - res := regexp.MustCompile(`\[(.*)\]$`).FindStringSubmatch(string(buf)) + //It can happen that the end of the "real" output doesn't not contain a new line + //that's why looking for the end isn't an option here (anymore?) + res := regexp.MustCompile(`(?m)^\[(.*)\]`).FindStringSubmatch(string(buf)) if len(res) == 0 { return []CephBlockImage{}, nil - } else { - buf = []byte(res[0]) } + buf = []byte(res[0]) var images []CephBlockImage - err = json.Unmarshal(buf, &images) - if err != nil { - return nil, fmt.Errorf("unmarshal failed: %+v. raw buffer response: %s", err, string(buf)) + if err = json.Unmarshal(buf, &images); err != nil { + return nil, fmt.Errorf("unmarshal failed: %+v. raw buffer response: %s", err, string(buf)) } return images, nil diff --git a/pkg/ceph/mon/config.go b/pkg/ceph/mon/config.go index 2a8424ea6..730484a57 100644 --- a/pkg/ceph/mon/config.go +++ b/pkg/ceph/mon/config.go @@ -102,8 +102,7 @@ func GenerateAdminConnectionConfig(context *clusterd.Context, cluster *ClusterIn return fmt.Errorf("failed to write keyring to %s. %+v", root, err) } - _, err = GenerateConfigFile(context, cluster, root, client.AdminUsername, keyringPath, false, nil, nil) - if err != nil { + if _, err = GenerateConfigFile(context, cluster, root, client.AdminUsername, keyringPath, false, nil, nil); err != nil { return fmt.Errorf("failed to write config to %s. %+v", root, err) } logger.Infof("generated admin config in %s", root) @@ -251,7 +250,7 @@ func CreateDefaultCephConfig(context *clusterd.Context, cluster *ClusterInfo, ru cephLogLevel := logLevelToCephLogLevel(context.LogLevel) return &cephConfig{ - &GlobalConfig{ + GlobalConfig: &GlobalConfig{ EnableExperimental: experimental, FSID: cluster.FSID, RunDir: runDir, diff --git a/pkg/ceph/mon/daemon.go b/pkg/ceph/mon/daemon.go index d3ed028af..e2d4b430a 100644 --- a/pkg/ceph/mon/daemon.go +++ b/pkg/ceph/mon/daemon.go @@ -25,17 +25,18 @@ import ( ) const ( - Port = 6790 + DefaultPort = 6790 ) type Config struct { Name string Cluster *ClusterInfo isDaemon bool + Port int32 } -func NewConfig(name string, cluster *ClusterInfo, isDaemon bool) *Config { - return &Config{Name: name, Cluster: cluster, isDaemon: isDaemon} +func NewConfig(name string, cluster *ClusterInfo, isDaemon bool, port int32) *Config { + return &Config{Name: name, Cluster: cluster, isDaemon: isDaemon, Port: port} } func FlattenMonEndpoints(mons map[string]*CephMonitorConfig) string { @@ -61,8 +62,8 @@ func ParseMonEndpoints(input string) map[string]*CephMonitorConfig { return mons } -func ToCephMon(name, ip string) *CephMonitorConfig { - return &CephMonitorConfig{Name: name, Endpoint: fmt.Sprintf("%s:%d", ip, Port)} +func ToCephMon(name, ip string, port int32) *CephMonitorConfig { + return &CephMonitorConfig{Name: name, Endpoint: fmt.Sprintf("%s:%d", ip, port)} } func Run(context *clusterd.Context, config *Config) error { @@ -150,8 +151,8 @@ func startMon(context *clusterd.Context, config *Config, confFilePath, monDataDi fmt.Sprintf("--mon-data=%s", monDataDir), fmt.Sprintf("--conf=%s", confFilePath), fmt.Sprintf("--keyring=%s", keyringPath), - fmt.Sprintf("--public-addr=%s:%d", context.NetworkInfo.PublicAddrIPv4, Port), - fmt.Sprintf("--public-bind-addr=%s:%d", context.NetworkInfo.ClusterAddrIPv4, Port), + fmt.Sprintf("--public-addr=%s:%d", context.NetworkInfo.PublicAddrIPv4, config.Port), + fmt.Sprintf("--public-bind-addr=%s:%d", context.NetworkInfo.ClusterAddrIPv4, config.Port), } err = context.ProcMan.Run(config.Name, "ceph-mon", args...) if err != nil { diff --git a/pkg/ceph/mon/leader.go b/pkg/ceph/mon/leader.go index 2abc55840..4788720e0 100644 --- a/pkg/ceph/mon/leader.go +++ b/pkg/ceph/mon/leader.go @@ -33,7 +33,8 @@ import ( ) const ( - monitorKey = "monitor" + monitorKey = "monitor" + UnhealthyHeartbeatAgeSeconds = 10 ) @@ -45,7 +46,7 @@ func NewLeader() *Leader { return &Leader{waitForQuorum: WaitForQuorum} } -// Apply the desired state to the cluster. The context provides all the information needed to make changes to the service. +// Configure the desired state to the cluster. The context provides all the information needed to make changes to the service. // Create the ceph monitors // Must be idempotent func (m *Leader) Configure(context *clusterd.Context, adminSecret string) error { @@ -68,13 +69,12 @@ func (m *Leader) Configure(context *clusterd.Context, adminSecret string) error } // trigger the monitors to start on each node - err = clusterd.TriggerAgentsAndWaitForCompletion(context.EtcdClient, monIDs(cluster.Monitors), monitorAgentName, len(cluster.Monitors)) - if err != nil { + if err = clusterd.TriggerAgentsAndWaitForCompletion(context.EtcdClient, monIDs(cluster.Monitors), monitorAgentName, len(cluster.Monitors)); err != nil { return err } // write the latest config to the config dir - if err := GenerateAdminConnectionConfig(context, cluster); err != nil { + if err = GenerateAdminConnectionConfig(context, cluster); err != nil { return fmt.Errorf("failed to write connection config for new mons. %+v", err) } @@ -95,8 +95,7 @@ func (m *Leader) Configure(context *clusterd.Context, adminSecret string) error } // notify the quorum to remove the bad members - err = removeMonitorsFromQuorum(context, cluster, monitorsToRemove) - if err != nil { + if err = removeMonitorsFromQuorum(context, cluster, monitorsToRemove); err != nil { logger.Warningf("failed to remove monitors from quorum. %v", err) } @@ -124,8 +123,7 @@ func removeMonitorsFromQuorum(context *clusterd.Context, cluster *ClusterInfo, m func RemoveMonitorFromQuorum(context *clusterd.Context, clusterName, name string) error { logger.Debugf("removing monitor %s", name) args := []string{"mon", "remove", name} - _, err := client.ExecuteCephCommand(context, clusterName, args) - if err != nil { + if _, err := client.ExecuteCephCommand(context, clusterName, args); err != nil { return fmt.Errorf("mon %s remove failed: %+v", name, err) } @@ -265,9 +263,9 @@ func chooseMonitorNodes(context *clusterd.Context) (map[string]*CephMonitorConfi monitorID := fmt.Sprintf("mon%d", nextMonID) settings[path.Join(nodeID, "id")] = monitorID settings[path.Join(nodeID, "ipaddress")] = node.PublicIP - settings[path.Join(nodeID, "port")] = strconv.Itoa(Port) + settings[path.Join(nodeID, "port")] = strconv.Itoa(int(DefaultPort)) - monitor := &CephMonitorConfig{Name: monitorID, Endpoint: fmt.Sprintf("%s:%d", node.PublicIP, Port)} + monitor := &CephMonitorConfig{Name: monitorID, Endpoint: fmt.Sprintf("%s:%d", node.PublicIP, DefaultPort)} monitors[nodeID] = monitor nextMonID++ diff --git a/pkg/operator/api/api.go b/pkg/operator/api/api.go index 66c08db8d..29a5df222 100644 --- a/pkg/operator/api/api.go +++ b/pkg/operator/api/api.go @@ -69,21 +69,23 @@ var clusterAccessRules = []v1beta1.PolicyRule{ // Cluster has the api service properties type Cluster struct { - context *clusterd.Context - Namespace string - placement k8sutil.Placement - Version string - Replicas int32 + context *clusterd.Context + Namespace string + placement k8sutil.Placement + Version string + Replicas int32 + HostNetwork bool } // New creates an instance -func New(context *clusterd.Context, namespace, version string, placement k8sutil.Placement) *Cluster { +func New(context *clusterd.Context, namespace, version string, placement k8sutil.Placement, hostNetwork bool) *Cluster { return &Cluster{ - context: context, - Namespace: namespace, - placement: placement, - Version: version, - Replicas: 1, + context: context, + Namespace: namespace, + placement: placement, + Version: version, + Replicas: 1, + HostNetwork: hostNetwork, } } @@ -168,6 +170,10 @@ func (c *Cluster) makeDeployment() *extensions.Deployment { Volumes: []v1.Volume{ {Name: k8sutil.DataDirVolume, VolumeSource: v1.VolumeSource{EmptyDir: &v1.EmptyDirVolumeSource{}}}, }, + HostNetwork: c.HostNetwork, + } + if c.HostNetwork { + podSpec.DNSPolicy = v1.DNSClusterFirstWithHostNet } c.placement.ApplyToPodSpec(&podSpec) @@ -192,6 +198,7 @@ func (c *Cluster) apiContainer() v1.Container { "api", fmt.Sprintf("--config-dir=%s", k8sutil.DataDir), fmt.Sprintf("--port=%d", model.Port), + fmt.Sprintf("--hostnetwork=%t", c.HostNetwork), }, Name: deploymentName, Image: k8sutil.MakeRookImage(c.Version), diff --git a/pkg/operator/api/api_test.go b/pkg/operator/api/api_test.go index c01adc678..19e730594 100644 --- a/pkg/operator/api/api_test.go +++ b/pkg/operator/api/api_test.go @@ -31,7 +31,7 @@ import ( func TestStartAPI(t *testing.T) { clientset := testop.New(3) - c := New(&clusterd.Context{Clientset: clientset}, "ns", "myversion", k8sutil.Placement{}) + c := New(&clusterd.Context{Clientset: clientset}, "ns", "myversion", k8sutil.Placement{}, false) // start a basic cluster err := c.Start() @@ -59,7 +59,7 @@ func validateStart(t *testing.T, c *Cluster) { func TestPodSpecs(t *testing.T) { clientset := testop.New(1) - c := New(&clusterd.Context{Clientset: clientset}, "ns", "myversion", k8sutil.Placement{}) + c := New(&clusterd.Context{Clientset: clientset}, "ns", "myversion", k8sutil.Placement{}, false) d := c.makeDeployment() assert.NotNil(t, d) @@ -88,7 +88,7 @@ func TestPodSpecs(t *testing.T) { func TestClusterRole(t *testing.T) { clientset := testop.New(1) - c := New(&clusterd.Context{Clientset: clientset}, "ns", "myversion", k8sutil.Placement{}) + c := New(&clusterd.Context{Clientset: clientset}, "ns", "myversion", k8sutil.Placement{}, false) // the role is create err := c.makeClusterRole() diff --git a/pkg/operator/cluster/controller.go b/pkg/operator/cluster/controller.go index ea56e610e..f5bc300bd 100644 --- a/pkg/operator/cluster/controller.go +++ b/pkg/operator/cluster/controller.go @@ -164,7 +164,7 @@ func (c *Cluster) createInstance() error { } // Start the mon pods - c.mons = mon.New(c.context, c.Namespace, c.Spec.DataDirHostPath, c.Spec.VersionTag, c.Spec.Placement.GetMON()) + c.mons = mon.New(c.context, c.Namespace, c.Spec.DataDirHostPath, c.Spec.VersionTag, c.Spec.Placement.GetMON(), c.Spec.HostNetwork) clusterInfo, err := c.mons.Start() if err != nil { return fmt.Errorf("failed to start the mons. %+v", err) @@ -175,20 +175,20 @@ func (c *Cluster) createInstance() error { return fmt.Errorf("failed to create initial crushmap: %+v", err) } - c.mgrs = mgr.New(c.context, c.Namespace, c.Spec.VersionTag, c.Spec.Placement.GetMGR()) + c.mgrs = mgr.New(c.context, c.Namespace, c.Spec.VersionTag, c.Spec.Placement.GetMGR(), c.Spec.HostNetwork) err = c.mgrs.Start() if err != nil { return fmt.Errorf("failed to start the ceph mgr. %+v", err) } - c.apis = api.New(c.context, c.Namespace, c.Spec.VersionTag, c.Spec.Placement.GetAPI()) + c.apis = api.New(c.context, c.Namespace, c.Spec.VersionTag, c.Spec.Placement.GetAPI(), c.Spec.HostNetwork) err = c.apis.Start() if err != nil { return fmt.Errorf("failed to start the REST api. %+v", err) } // Start the OSDs - c.osds = osd.New(c.context, c.Namespace, c.Spec.VersionTag, c.Spec.Storage, c.Spec.DataDirHostPath, c.Spec.Placement.GetOSD()) + c.osds = osd.New(c.context, c.Namespace, c.Spec.VersionTag, c.Spec.Storage, c.Spec.DataDirHostPath, c.Spec.Placement.GetOSD(), c.Spec.HostNetwork) err = c.osds.Start() if err != nil { return fmt.Errorf("failed to start the osds. %+v", err) diff --git a/pkg/operator/cluster/types.go b/pkg/operator/cluster/types.go index f522b3d52..850b98375 100644 --- a/pkg/operator/cluster/types.go +++ b/pkg/operator/cluster/types.go @@ -76,6 +76,9 @@ type ClusterSpec struct { // A spec for available storage in the cluster and how it should be used Storage osd.StorageSpec `json:"storage"` + + // HostNetwork to enable host network + HostNetwork bool `json:"hostNetwork"` } // PlacementSpec is a set of Placement configurations for the rook cluster. diff --git a/pkg/operator/mds/mds.go b/pkg/operator/mds/mds.go index 938a632e1..e0189db88 100644 --- a/pkg/operator/mds/mds.go +++ b/pkg/operator/mds/mds.go @@ -43,23 +43,25 @@ const ( // Cluster for mds management type Cluster struct { - Namespace string - Version string - Replicas int32 - context *clusterd.Context - dataDir string - placement k8sutil.Placement + Namespace string + Version string + Replicas int32 + context *clusterd.Context + dataDir string + placement k8sutil.Placement + HostNetwork bool } // New creates an instance of the mds manager -func New(context *clusterd.Context, namespace, version string, placement k8sutil.Placement) *Cluster { +func New(context *clusterd.Context, namespace, version string, placement k8sutil.Placement, hostNetwork bool) *Cluster { return &Cluster{ - context: context, - Namespace: namespace, - placement: placement, - Version: version, - Replicas: 1, - dataDir: k8sutil.DataDir, + context: context, + Namespace: namespace, + placement: placement, + Version: version, + Replicas: 1, + dataDir: k8sutil.DataDir, + HostNetwork: hostNetwork, } } @@ -133,6 +135,10 @@ func (c *Cluster) makeDeployment(id string) *extensions.Deployment { {Name: k8sutil.DataDirVolume, VolumeSource: v1.VolumeSource{EmptyDir: &v1.EmptyDirVolumeSource{}}}, k8sutil.ConfigOverrideVolume(), }, + HostNetwork: c.HostNetwork, + } + if c.HostNetwork { + podSpec.DNSPolicy = v1.DNSClusterFirstWithHostNet } c.placement.ApplyToPodSpec(&podSpec) diff --git a/pkg/operator/mds/mds_test.go b/pkg/operator/mds/mds_test.go index b1411297d..ed2d95d1f 100644 --- a/pkg/operator/mds/mds_test.go +++ b/pkg/operator/mds/mds_test.go @@ -42,7 +42,7 @@ func TestStartMDS(t *testing.T) { Executor: executor, ConfigDir: configDir, Clientset: testop.New(3)} - c := New(context, "ns", "myversion", k8sutil.Placement{}) + c := New(context, "ns", "myversion", k8sutil.Placement{}, false) defer os.RemoveAll(c.dataDir) // start a basic cluster @@ -64,7 +64,7 @@ func validateStart(t *testing.T, c *Cluster) { } func TestPodSpecs(t *testing.T) { - c := New(nil, "ns", "myversion", k8sutil.Placement{}) + c := New(nil, "ns", "myversion", k8sutil.Placement{}, false) mdsID := "mds1" d := c.makeDeployment(mdsID) @@ -87,3 +87,13 @@ func TestPodSpecs(t *testing.T) { assert.Equal(t, "--config-dir=/var/lib/rook", cont.Args[1]) assert.Equal(t, "--mds-id=mds1", cont.Args[2]) } + +func TestHostNetwork(t *testing.T) { + c := New(nil, "ns", "myversion", k8sutil.Placement{}, true) + mdsID := "mds1" + + d := c.makeDeployment(mdsID) + + assert.Equal(t, true, d.Spec.Template.Spec.HostNetwork) + assert.Equal(t, v1.DNSClusterFirstWithHostNet, d.Spec.Template.Spec.DNSPolicy) +} diff --git a/pkg/operator/mgr/mgr.go b/pkg/operator/mgr/mgr.go index c2500cf10..70f1c1188 100644 --- a/pkg/operator/mgr/mgr.go +++ b/pkg/operator/mgr/mgr.go @@ -40,23 +40,25 @@ const ( // Cluster is the ceph mgr manager type Cluster struct { - Namespace string - Version string - Replicas int - placement k8sutil.Placement - context *clusterd.Context - dataDir string + Namespace string + Version string + Replicas int + placement k8sutil.Placement + context *clusterd.Context + dataDir string + HostNetwork bool } // New creates an instance of the mgr -func New(context *clusterd.Context, namespace, version string, placement k8sutil.Placement) *Cluster { +func New(context *clusterd.Context, namespace, version string, placement k8sutil.Placement, hostNetwork bool) *Cluster { return &Cluster{ - context: context, - Namespace: namespace, - placement: placement, - Version: version, - Replicas: 1, - dataDir: k8sutil.DataDir, + context: context, + Namespace: namespace, + placement: placement, + Version: version, + Replicas: 1, + dataDir: k8sutil.DataDir, + HostNetwork: hostNetwork, } } @@ -105,8 +107,12 @@ func (c *Cluster) makeDeployment(name string) *extensions.Deployment { {Name: k8sutil.DataDirVolume, VolumeSource: v1.VolumeSource{EmptyDir: &v1.EmptyDirVolumeSource{}}}, k8sutil.ConfigOverrideVolume(), }, + HostNetwork: c.HostNetwork, }, } + if c.HostNetwork { + podSpec.Spec.DNSPolicy = v1.DNSClusterFirstWithHostNet + } c.placement.ApplyToPodSpec(&podSpec.Spec) replicas := int32(1) diff --git a/pkg/operator/mgr/mgr_test.go b/pkg/operator/mgr/mgr_test.go index e32d3a506..aac55af98 100644 --- a/pkg/operator/mgr/mgr_test.go +++ b/pkg/operator/mgr/mgr_test.go @@ -43,7 +43,7 @@ func TestStartMGR(t *testing.T) { Executor: executor, ConfigDir: configDir, Clientset: testop.New(3)} - c := New(context, "ns", "myversion", k8sutil.Placement{}) + c := New(context, "ns", "myversion", k8sutil.Placement{}, false) defer os.RemoveAll(c.dataDir) // start a basic service @@ -68,7 +68,7 @@ func validateStart(t *testing.T, c *Cluster) { } func TestPodSpec(t *testing.T) { - c := New(nil, "ns", "myversion", k8sutil.Placement{}) + c := New(nil, "ns", "myversion", k8sutil.Placement{}, false) d := c.makeDeployment("mgr1") assert.NotNil(t, d) @@ -89,3 +89,13 @@ func TestPodSpec(t *testing.T) { assert.Equal(t, "mgr", cont.Args[0]) assert.Equal(t, "--config-dir=/var/lib/rook", cont.Args[1]) } + +func TestHostNetwork(t *testing.T) { + c := New(nil, "ns", "myversion", k8sutil.Placement{}, true) + + d := c.makeDeployment("mgr1") + assert.NotNil(t, d) + + assert.Equal(t, true, d.Spec.Template.Spec.HostNetwork) + assert.Equal(t, v1.DNSClusterFirstWithHostNet, d.Spec.Template.Spec.DNSPolicy) +} diff --git a/pkg/operator/mon/health.go b/pkg/operator/mon/health.go index 8158d3e3e..7bcbc27b0 100644 --- a/pkg/operator/mon/health.go +++ b/pkg/operator/mon/health.go @@ -125,7 +125,7 @@ func (c *Cluster) failoverMon(name string) error { logger.Infof("Failing over monitor %s", name) // Start a new monitor - m := &monConfig{Name: fmt.Sprintf("%s%d", appName, c.maxMonID+1), Port: int32(mon.Port)} + m := &monConfig{Name: fmt.Sprintf("%s%d", appName, c.maxMonID+1), Port: int32(mon.DefaultPort)} logger.Infof("starting new mon %s", m.Name) // Create the service endpoint @@ -133,18 +133,21 @@ func (c *Cluster) failoverMon(name string) error { if err != nil { return fmt.Errorf("failed to create mon service. %+v", err) } - m.PublicIP = serviceIP - c.clusterInfo.Monitors[m.Name] = mon.ToCephMon(m.Name, m.PublicIP) - // Save the mon config - err = c.saveMonConfig() - if err != nil { - return fmt.Errorf("failed to save mons. %+v", err) + if !c.HostNetwork { + m.PublicIP = serviceIP + c.clusterInfo.Monitors[m.Name] = mon.ToCephMon(m.Name, m.PublicIP, m.Port) + } + + mConf := []*monConfig{m} + + // Assign the pod to a node + if err = c.assignMons(mConf); err != nil { + return fmt.Errorf("failed to assign pods to mons. %+v", err) } // Start the pod - err = c.startPods([]*monConfig{m}) - if err != nil { + if err = c.startPods(mConf); err != nil { return fmt.Errorf("failed to start new mon %s. %+v", m.Name, err) } @@ -161,8 +164,7 @@ func (c *Cluster) removeMon(name string) error { var gracePeriod int64 propagation := metav1.DeletePropagationForeground options := &metav1.DeleteOptions{GracePeriodSeconds: &gracePeriod, PropagationPolicy: &propagation} - err := c.context.Clientset.Extensions().ReplicaSets(c.Namespace).Delete(name, options) - if err != nil { + if err := c.context.Clientset.Extensions().ReplicaSets(c.Namespace).Delete(name, options); err != nil { if errors.IsNotFound(err) { logger.Infof("dead mon %s was already gone", name) } else { @@ -171,15 +173,26 @@ func (c *Cluster) removeMon(name string) error { } // Remove the bad monitor from quorum - err = mon.RemoveMonitorFromQuorum(c.context, c.clusterInfo.Name, name) - if err != nil { + if err := mon.RemoveMonitorFromQuorum(c.context, c.clusterInfo.Name, name); err != nil { return fmt.Errorf("failed to remove mon %s from quorum. %+v", name, err) } delete(c.clusterInfo.Monitors, name) + nodeName := c.mapping.Node[name].Name + // if node->port "mapping" has been created, decrease or delete it + if port, ok := c.mapping.Port[nodeName]; ok { + if port == mon.DefaultPort { + delete(c.mapping.Port, nodeName) + } + // don't clean up if a node port is higher than the default port, other + //mons could be on the same node with > DefaultPort ports, decreasing could + //cause port collsions + // This can be solved by using a map[nodeName][]int32 for the ports to + //even better check which ports are open for the HostNetwork mode + } + delete(c.mapping.Node, name) // Remove the service endpoint - err = c.context.Clientset.CoreV1().Services(c.Namespace).Delete(name, options) - if err != nil { + if err := c.context.Clientset.CoreV1().Services(c.Namespace).Delete(name, options); err != nil { if errors.IsNotFound(err) { logger.Infof("dead mon service %s was already gone", name) } else { @@ -187,10 +200,14 @@ func (c *Cluster) removeMon(name string) error { } } - err = c.saveMonConfig() - if err != nil { + if err := c.saveMonConfig(); err != nil { return fmt.Errorf("failed to save mon config after failing over mon %s. %+v", name, err) } + // make sure to rewrite the config so NO new connections are made to the removed mon + if err := c.writeConnectionConfig(); err != nil { + return fmt.Errorf("failed to write connection config after failing over mon %s. %+v", name, err) + } + return nil } diff --git a/pkg/operator/mon/mon.go b/pkg/operator/mon/mon.go index a2ccbe371..fe106b0f6 100644 --- a/pkg/operator/mon/mon.go +++ b/pkg/operator/mon/mon.go @@ -18,6 +18,7 @@ limitations under the License. package mon import ( + "encoding/json" "fmt" "strconv" "strings" @@ -43,9 +44,12 @@ var logger = capnslog.NewPackageLogger("github.com/rook/rook", "op-mon") const ( // EndpointConfigMapName is the name of the configmap with mon endpoints EndpointConfigMapName = "rook-ceph-mon-endpoints" - // EndpointDataKey is the name of the key inside the mon configmap to get the endpoints EndpointDataKey = "data" + // MaxMonIDKey is the name of the max mon id used + MaxMonIDKey = "maxMonId" + // MappingKey is the name of the mapping for the mon->node and node->port + MappingKey = "mapping" appName = "rook-ceph-mon" monNodeAttr = "mon_node" @@ -55,7 +59,6 @@ const ( monSecretName = "mon-secret" adminSecretName = "admin-secret" clusterSecretName = "cluster-name" - maxMonIDKey = "maxMonId" ) // Cluster is for the cluster of monitors @@ -76,6 +79,8 @@ type Cluster struct { monPodRetryInterval time.Duration monPodTimeout time.Duration monTimeoutList map[string]time.Time + HostNetwork bool + mapping *mapping } // monConfig for a single monitor @@ -85,8 +90,19 @@ type monConfig struct { Port int32 } +// mapping mon node and port mapping +type mapping struct { + Node map[string]*nodeInfo `json:"node"` + Port map[string]int32 `json:"port"` +} + +type nodeInfo struct { + Name string + Address string +} + // New creates an instance of a mon cluster -func New(context *clusterd.Context, namespace, dataDirHostPath, version string, placement k8sutil.Placement) *Cluster { +func New(context *clusterd.Context, namespace, dataDirHostPath, version string, placement k8sutil.Placement, hostNetwork bool) *Cluster { return &Cluster{ context: context, placement: placement, @@ -99,6 +115,11 @@ func New(context *clusterd.Context, namespace, dataDirHostPath, version string, monPodRetryInterval: 6 * time.Second, monPodTimeout: 5 * time.Minute, monTimeoutList: map[string]time.Time{}, + HostNetwork: hostNetwork, + mapping: &mapping{ + Node: map[string]*nodeInfo{}, + Port: map[string]int32{}, + }, } } @@ -111,18 +132,37 @@ func (c *Cluster) Start() (*mon.ClusterInfo, error) { } if len(c.clusterInfo.Monitors) < c.Size { + // init the mons config + mons := c.initMonConfig(c.Size) + + // Assign the pods to nodes + if err := c.assignMons(mons); err != nil { + return nil, fmt.Errorf("failed to assign pods to mons. %+v", err) + } + // Start one monitor at a time for i := len(c.clusterInfo.Monitors); i < c.Size; i++ { - mons, err := c.initMonServices(i + 1) - if err != nil { - return nil, fmt.Errorf("failed to init mons. %+v", err) + // Init the mon IPs + if err := c.initMonIPs(mons[0 : i+1]); err != nil { + return nil, fmt.Errorf("failed to init mon services. %+v", err) } - err = c.startPods(mons) - if err != nil { + // save the mon config after we have "initiated the IPs" + if err := c.saveMonConfig(); err != nil { + return nil, fmt.Errorf("failed to save mons. %+v", err) + } + + // make sure we have the connection info generated so connections can happen + if err := c.writeConnectionConfig(); err != nil { + return nil, err + } + + // Start pods + if err := c.startPods(mons[0 : i+1]); err != nil { return nil, fmt.Errorf("failed to start mon pods. %+v", err) } } + logger.Debugf("mon endpoints used are: %s", c.clusterInfo.MonEndpoints()) } else { // Check the health of a previously started cluster if err := c.checkHealth(); err != nil { @@ -152,8 +192,7 @@ func (c *Cluster) initClusterInfo() error { return fmt.Errorf("failed to get mon secrets. %+v", err) } - err = c.createMonSecretsAndSave() - if err != nil { + if err = c.createMonSecretsAndSave(); err != nil { return err } } else { @@ -167,49 +206,55 @@ func (c *Cluster) initClusterInfo() error { } // get the existing monitor config - err = c.loadMonConfig() - if err != nil { + if err = c.loadMonConfig(); err != nil { return fmt.Errorf("failed to get mon config. %+v", err) } - // make sure we have the connection info generated so connections can happen - err = c.writeConnectionConfig() - if err != nil { + // make sure we have the connection info generated after loading it in the first mon init step + if err = c.writeConnectionConfig(); err != nil { return err } + return nil } -func (c *Cluster) initMonServices(size int) ([]*monConfig, error) { +func (c *Cluster) initMonConfig(size int) []*monConfig { mons := []*monConfig{} // initialize the mon pod info for mons that have been previously created for _, monitor := range c.clusterInfo.Monitors { - mons = append(mons, &monConfig{Name: monitor.Name, Port: int32(mon.Port)}) + mons = append(mons, &monConfig{Name: monitor.Name, Port: int32(mon.DefaultPort)}) } // initialize mon info if we don't have enough mons (at first startup) for i := len(c.clusterInfo.Monitors); i < size; i++ { c.maxMonID++ - mons = append(mons, &monConfig{Name: fmt.Sprintf("%s%d", appName, c.maxMonID), Port: int32(mon.Port)}) + mons = append(mons, &monConfig{Name: fmt.Sprintf("%s%d", appName, c.maxMonID), Port: int32(mon.DefaultPort)}) } + return mons +} + +func (c *Cluster) initMonIPs(mons []*monConfig) error { for _, m := range mons { - serviceIP, err := c.createService(m) - if err != nil { - return nil, fmt.Errorf("failed to create mon service. %+v", err) + if c.HostNetwork { + logger.Infof("setting mon endpoints for hostnetwork mode") + node, ok := c.mapping.Node[m.Name] + if !ok { + return fmt.Errorf("mon doesn't exit in assignment map") + } + m.PublicIP = node.Address + } else { + serviceIP, err := c.createService(m) + if err != nil { + return fmt.Errorf("failed to create mon service. %+v", err) + } + m.PublicIP = serviceIP } - m.PublicIP = serviceIP - c.clusterInfo.Monitors[m.Name] = mon.ToCephMon(m.Name, m.PublicIP) + c.clusterInfo.Monitors[m.Name] = mon.ToCephMon(m.Name, m.PublicIP, m.Port) } - // save the mon config - err := c.saveMonConfig() - if err != nil { - return nil, fmt.Errorf("failed to save mons. %+v", err) - } - - return mons, nil + return nil } // get the ID of a monitor from its name @@ -290,6 +335,9 @@ func (c *Cluster) createService(mon *monConfig) (string, error) { Selector: labels, }, } + if c.HostNetwork { + s.Spec.ClusterIP = v1.ClusterIPNone + } s, err := c.context.Clientset.CoreV1().Services(c.Namespace).Create(s) if err != nil { @@ -306,11 +354,12 @@ func (c *Cluster) createService(mon *monConfig) (string, error) { logger.Warningf("service ip not found for mon %s. this better be a test", mon.Name) return "", nil } + logger.Infof("mon %s running at %s:%d", mon.Name, s.Spec.ClusterIP, mon.Port) return s.Spec.ClusterIP, nil } -func (c *Cluster) startPods(mons []*monConfig) error { +func (c *Cluster) assignMons(mons []*monConfig) error { // schedule the mons on different nodes if we have enough nodes to be unique availableNodes, err := c.getAvailableMonNodes() if err != nil { @@ -319,12 +368,60 @@ func (c *Cluster) startPods(mons []*monConfig) error { nodeIndex := 0 for _, m := range mons { + if _, ok := c.mapping.Node[m.Name]; ok { + logger.Debugf("mon %s already assigned to a node, no need to assign", m.Name) + continue + } + + // when the available nodes are >= than the mon size we just begin taking + // "node0" through nodeX (X = c.Size-1) + var node v1.Node // pick one of the available nodes where the mon will be assigned - node := availableNodes[nodeIndex%len(availableNodes)] + node = availableNodes[nodeIndex%len(availableNodes)] + logger.Debugf("mon %s assigned to node %s", m.Name, node.Name) + nodeInfo, err := getNodeInfoFromNode(node) + if err != nil { + return fmt.Errorf("couldn't get node info from node %s. %+v", node.Name, err) + } + // when hostNetwork is used check if we need to increase the port of the node + if c.HostNetwork { + if _, ok := c.mapping.Port[node.Name]; ok { + // when the node was already chosen increase port by 1 and set + // assignment and that the node was chosen + m.Port = c.mapping.Port[node.Name] + int32(1) + } + c.mapping.Port[node.Name] = m.Port + } + c.mapping.Node[m.Name] = nodeInfo nodeIndex++ + } + + logger.Debug("assigned mons to nodes") + return nil +} + +func getNodeInfoFromNode(n v1.Node) (*nodeInfo, error) { + nr := &nodeInfo{} + nr.Name = n.Name + for _, ip := range n.Status.Addresses { + if ip.Type == v1.NodeExternalIP || ip.Type == v1.NodeInternalIP { + logger.Debugf("using IP %s for node %s", ip.Address, n.Name) + nr.Address = ip.Address + break + } + } + if nr.Address == "" { + return nil, fmt.Errorf("no IP given for node %s", nr.Name) + } + return nr, nil +} + +func (c *Cluster) startPods(mons []*monConfig) error { + for _, m := range mons { + node, _ := c.mapping.Node[m.Name] // start the mon replicaset/pod - err = c.startMon(m, node.Name) + err := c.startMon(m, node.Name) if err != nil { return fmt.Errorf("failed to create pod %s. %+v", m.Name, err) } @@ -361,13 +458,19 @@ func (c *Cluster) saveMonConfig() error { Annotations: map[string]string{}, }, } - configMap.Data = map[string]string{ - EndpointDataKey: mon.FlattenMonEndpoints(c.clusterInfo.Monitors), - maxMonIDKey: strconv.Itoa(c.maxMonID), + + monMapping, err := json.Marshal(c.mapping) + if err != nil { + return fmt.Errorf("failed to marshal mon mapping. %+v", err) } - _, err := c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Create(configMap) - if err != nil { + configMap.Data = map[string]string{ + EndpointDataKey: mon.FlattenMonEndpoints(c.clusterInfo.Monitors), + MaxMonIDKey: strconv.Itoa(c.maxMonID), + MappingKey: string(monMapping), + } + + if _, err := c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Create(configMap); err != nil { if !errors.IsAlreadyExists(err) { return fmt.Errorf("failed to create mon endpoint config map. %+v", err) } @@ -404,8 +507,12 @@ func (c *Cluster) loadMonConfig() error { return err } // If the config map was not found, initialize the empty set of monitors - c.maxMonID = -1 + c.mapping = &mapping{ + Node: map[string]*nodeInfo{}, + Port: map[string]int32{}, + } c.clusterInfo.Monitors = map[string]*mon.CephMonitorConfig{} + c.maxMonID = -1 return c.saveMonConfig() } @@ -417,13 +524,29 @@ func (c *Cluster) loadMonConfig() error { } // Parse the max monitor id - if id, ok := cm.Data[maxMonIDKey]; ok { + if id, ok := cm.Data[MaxMonIDKey]; ok { c.maxMonID, err = strconv.Atoi(id) if err != nil { logger.Errorf("invalid max mon id %s. %+v", id, err) } } + // Unmarshal mon mapping (node, port) + if mappingStr, ok := cm.Data[MappingKey]; ok && mappingStr != "" { + if err := json.Unmarshal([]byte(mappingStr), &c.mapping); err != nil { + logger.Errorf("invalid mapping json. json=%s; %+v", mappingStr, err) + c.mapping = &mapping{ + Node: map[string]*nodeInfo{}, + Port: map[string]int32{}, + } + } + } else { + c.mapping = &mapping{ + Node: map[string]*nodeInfo{}, + Port: map[string]int32{}, + } + } + // Make sure the max id is consistent with the current monitors for _, m := range c.clusterInfo.Monitors { id, _ := getMonID(m.Name) @@ -432,7 +555,7 @@ func (c *Cluster) loadMonConfig() error { } } - logger.Infof("loaded: maxMonID=%d, mons=%+v", c.maxMonID, c.clusterInfo.Monitors) + logger.Infof("loaded: maxMonID=%d, mons=%+v mapping=%+v", c.maxMonID, c.clusterInfo.Monitors, c.mapping) return nil } diff --git a/pkg/operator/mon/mon_test.go b/pkg/operator/mon/mon_test.go index 61282cba2..0d8ca5eba 100644 --- a/pkg/operator/mon/mon_test.go +++ b/pkg/operator/mon/mon_test.go @@ -19,6 +19,7 @@ import ( "fmt" "io/ioutil" "path" + "strconv" "strings" "testing" "time" @@ -27,6 +28,7 @@ import ( "github.com/rook/rook/pkg/ceph/client" clienttest "github.com/rook/rook/pkg/ceph/client/test" + cephmon "github.com/rook/rook/pkg/ceph/mon" cephtest "github.com/rook/rook/pkg/ceph/test" "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/operator/k8sutil" @@ -37,9 +39,8 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) -func TestStartMonPods(t *testing.T) { +func newTestStartCluster(namespace string) *clusterd.Context { clientset := test.New(3) - namespace := "ns" configDir, _ := ioutil.TempDir("", "") defer os.RemoveAll(configDir) executor := &exectest.MockExecutor{ @@ -50,12 +51,16 @@ func TestStartMonPods(t *testing.T) { return "", nil }, } - context := &clusterd.Context{ + return &clusterd.Context{ Clientset: clientset, Executor: executor, ConfigDir: configDir, } - c := &Cluster{ +} + +func newCluster(context *clusterd.Context, namespace string, hostNetwork bool) *Cluster { + return &Cluster{ + HostNetwork: true, context: context, Namespace: namespace, Version: "myversion", @@ -64,7 +69,18 @@ func TestStartMonPods(t *testing.T) { waitForStart: false, monPodRetryInterval: 10 * time.Millisecond, monPodTimeout: 1 * time.Second, + monTimeoutList: map[string]time.Time{}, + mapping: &mapping{ + Node: map[string]*nodeInfo{}, + Port: map[string]int32{}, + }, } +} + +func TestStartMonPods(t *testing.T) { + namespace := "ns" + context := newTestStartCluster(namespace) + c := newCluster(context, namespace, false) // start a basic cluster info, err := c.Start() @@ -81,6 +97,51 @@ func TestStartMonPods(t *testing.T) { validateStart(t, c) } +func TestOperatorRestart(t *testing.T) { + namespace := "ns" + context := newTestStartCluster(namespace) + c := newCluster(context, namespace, false) + + // start a basic cluster + info, err := c.Start() + assert.Nil(t, err) + assert.NotNil(t, info) + + validateStart(t, c) + + c = newCluster(context, namespace, false) + + // starting again should be a no-op, but still results in an error + info, err = c.Start() + assert.Nil(t, err) + assert.NotNil(t, info) + + validateStart(t, c) +} + +// safety check that if hostNetwork is used no changes occur on an operator restart +func TestOperatorRestartHostNetwork(t *testing.T) { + namespace := "ns" + context := newTestStartCluster(namespace) + c := newCluster(context, namespace, true) + + // start a basic cluster + info, err := c.Start() + assert.Nil(t, err) + assert.NotNil(t, info) + + validateStart(t, c) + + c = newCluster(context, namespace, true) + + // starting again should be a no-op, but still results in an error + info, err = c.Start() + assert.Nil(t, err) + assert.NotNil(t, info) + + validateStart(t, c) +} + func validateStart(t *testing.T, c *Cluster) { s, err := c.context.Clientset.CoreV1().Secrets(c.Namespace).Get("rook-admin", metav1.GetOptions{}) assert.Nil(t, err) @@ -99,7 +160,7 @@ func TestSaveMonEndpoints(t *testing.T) { clientset := test.New(1) configDir, _ := ioutil.TempDir("", "") defer os.RemoveAll(configDir) - c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", "", "myversion", k8sutil.Placement{}) + c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", "", "myversion", k8sutil.Placement{}, false) c.clusterInfo = test.CreateClusterInfo(1) // create the initial config map @@ -109,15 +170,25 @@ func TestSaveMonEndpoints(t *testing.T) { cm, err := c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(EndpointConfigMapName, metav1.GetOptions{}) assert.Nil(t, err) assert.Equal(t, "mon1=1.2.3.1:6790", cm.Data[EndpointDataKey]) + assert.Equal(t, `{"node":{},"port":{}}`, cm.Data[MappingKey]) + assert.Equal(t, "-1", cm.Data[MaxMonIDKey]) // update the config map c.clusterInfo.Monitors["mon1"].Endpoint = "2.3.4.5:6790" + c.maxMonID = 2 + c.mapping.Node["mon1"] = &nodeInfo{ + Name: "node0", + Address: "1.1.1.1", + } + c.mapping.Port["node0"] = int32(12345) err = c.saveMonConfig() assert.Nil(t, err) cm, err = c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(EndpointConfigMapName, metav1.GetOptions{}) assert.Nil(t, err) assert.Equal(t, "mon1=2.3.4.5:6790", cm.Data[EndpointDataKey]) + assert.Equal(t, `{"node":{"mon1":{"Name":"node0","Address":"1.1.1.1"}},"port":{"node0":12345}}`, cm.Data[MappingKey]) + assert.Equal(t, "2", cm.Data[MaxMonIDKey]) } func TestCheckHealth(t *testing.T) { @@ -134,7 +205,7 @@ func TestCheckHealth(t *testing.T) { ConfigDir: configDir, Executor: executor, } - c := New(context, "ns", "", "myversion", k8sutil.Placement{}) + c := New(context, "ns", "", "myversion", k8sutil.Placement{}, false) c.clusterInfo = test.CreateClusterInfo(1) c.waitForStart = false defer os.RemoveAll(c.context.ConfigDir) @@ -142,6 +213,12 @@ func TestCheckHealth(t *testing.T) { err := c.checkHealth() assert.Nil(t, err) + c.mapping.Node["mon1"] = &nodeInfo{ + Name: "node0", + Address: "", + } + c.mapping.Port["node0"] = cephmon.DefaultPort + c.maxMonID = 10 err = c.failoverMon("mon1") assert.Nil(t, err) @@ -191,7 +268,7 @@ func TestMonID(t *testing.T) { func TestAvailableMonNodes(t *testing.T) { clientset := test.New(1) - c := New(&clusterd.Context{Clientset: clientset}, "ns", "", "myversion", k8sutil.Placement{}) + c := New(&clusterd.Context{Clientset: clientset}, "ns", "", "myversion", k8sutil.Placement{}, false) c.clusterInfo = test.CreateClusterInfo(0) nodes, err := c.getAvailableMonNodes() assert.Nil(t, err) @@ -208,7 +285,7 @@ func TestAvailableMonNodes(t *testing.T) { func TestAvailableNodesInUse(t *testing.T) { clientset := test.New(3) - c := New(&clusterd.Context{Clientset: clientset}, "ns", "", "myversion", k8sutil.Placement{}) + c := New(&clusterd.Context{Clientset: clientset}, "ns", "", "myversion", k8sutil.Placement{}, false) c.clusterInfo = test.CreateClusterInfo(0) // all three nodes are available by default @@ -219,7 +296,7 @@ func TestAvailableNodesInUse(t *testing.T) { // start pods on two of the nodes so that only one node will be available for i := 0; i < 2; i++ { pod := c.makeMonPod(&monConfig{Name: fmt.Sprintf("rook-ceph-mon%d", i)}, nodes[i].Name) - _, err := clientset.CoreV1().Pods(c.Namespace).Create(pod) + _, err = clientset.CoreV1().Pods(c.Namespace).Create(pod) assert.Nil(t, err) } reducedNodes, err := c.getAvailableMonNodes() @@ -239,7 +316,7 @@ func TestAvailableNodesInUse(t *testing.T) { func TestTaintedNodes(t *testing.T) { clientset := test.New(3) - c := New(&clusterd.Context{Clientset: clientset}, "ns", "", "myversion", k8sutil.Placement{}) + c := New(&clusterd.Context{Clientset: clientset}, "ns", "", "myversion", k8sutil.Placement{}, false) c.clusterInfo = test.CreateClusterInfo(0) nodes, err := c.getAvailableMonNodes() @@ -272,7 +349,7 @@ func TestTaintedNodes(t *testing.T) { func TestNodeAffinity(t *testing.T) { clientset := test.New(3) - c := New(&clusterd.Context{Clientset: clientset}, "ns", "", "myversion", k8sutil.Placement{}) + c := New(&clusterd.Context{Clientset: clientset}, "ns", "", "myversion", k8sutil.Placement{}, false) c.clusterInfo = test.CreateClusterInfo(0) nodes, err := c.getAvailableMonNodes() @@ -309,3 +386,93 @@ func TestNodeAffinity(t *testing.T) { assert.Equal(t, nodes[1].Name, cleanNodes[0].Name) assert.Equal(t, nodes[2].Name, cleanNodes[1].Name) } + +func TestHostNetwork(t *testing.T) { + clientset := test.New(3) + c := New(&clusterd.Context{Clientset: clientset}, "ns", "", "myversion", k8sutil.Placement{}, false) + c.clusterInfo = test.CreateClusterInfo(0) + + c.HostNetwork = true + + nodes, err := c.getAvailableMonNodes() + assert.Nil(t, err) + assert.Equal(t, 3, len(nodes)) + + pod := c.makeMonPod(&monConfig{Name: "mon2"}, nodes[2].Name) + assert.NotNil(t, pod) + + assert.Equal(t, true, pod.Spec.HostNetwork) + assert.Equal(t, v1.DNSClusterFirstWithHostNet, pod.Spec.DNSPolicy) +} + +func TestGetNodeInfoFromNode(t *testing.T) { + clientset := test.New(1) + node, err := clientset.CoreV1().Nodes().Get("node0", metav1.GetOptions{}) + assert.Nil(t, err) + assert.NotNil(t, node) + + node.Status = v1.NodeStatus{} + node.Status.Addresses = []v1.NodeAddress{ + { + Type: v1.NodeExternalIP, + Address: "1.1.1.1", + }, + } + + c := New(&clusterd.Context{Clientset: clientset}, "ns", "", "myversion", k8sutil.Placement{}, true) + c.clusterInfo = test.CreateClusterInfo(0) + + var info *nodeInfo + info, err = getNodeInfoFromNode(*node) + assert.Nil(t, err) + + assert.Equal(t, "1.1.1.1", info.Address) +} + +func TestHostNetworkPortIncrease(t *testing.T) { + clientset := test.New(1) + node, err := clientset.CoreV1().Nodes().Get("node0", metav1.GetOptions{}) + assert.Nil(t, err) + assert.NotNil(t, node) + + node.Status = v1.NodeStatus{} + node.Status.Addresses = []v1.NodeAddress{ + { + Type: v1.NodeExternalIP, + Address: "1.1.1.1", + }, + } + + configDir, _ := ioutil.TempDir("", "") + defer os.RemoveAll(configDir) + c := New(&clusterd.Context{ + Clientset: clientset, + ConfigDir: configDir, + }, "ns", "", "myversion", k8sutil.Placement{}, true) + c.clusterInfo = test.CreateClusterInfo(0) + + mons := []*monConfig{ + { + Name: "mon1", + Port: cephmon.DefaultPort, + }, + { + Name: "mon2", + Port: cephmon.DefaultPort, + }, + } + + err = c.assignMons(mons) + assert.Nil(t, err) + + err = c.initMonIPs(mons) + assert.Nil(t, err) + + assert.Equal(t, node.Name, c.mapping.Node["mon1"].Name) + assert.Equal(t, node.Name, c.mapping.Node["mon2"].Name) + + sEndpoint := strings.Split(c.clusterInfo.Monitors["mon1"].Endpoint, ":") + assert.Equal(t, strconv.Itoa(cephmon.DefaultPort), sEndpoint[1]) + sEndpoint = strings.Split(c.clusterInfo.Monitors["mon2"].Endpoint, ":") + assert.Equal(t, strconv.Itoa(cephmon.DefaultPort+1), sEndpoint[1]) +} diff --git a/pkg/operator/mon/spec.go b/pkg/operator/mon/spec.go index c7a078d0c..9c17d4b61 100644 --- a/pkg/operator/mon/spec.go +++ b/pkg/operator/mon/spec.go @@ -96,6 +96,10 @@ func (c *Cluster) makeMonPod(config *monConfig, nodeName string) *v1.Pod { {Name: k8sutil.DataDirVolume, VolumeSource: dataDirSource}, k8sutil.ConfigOverrideVolume(), }, + HostNetwork: c.HostNetwork, + } + if c.HostNetwork { + podSpec.DNSPolicy = v1.DNSClusterFirstWithHostNet } c.placement.ApplyToPodSpec(&podSpec) diff --git a/pkg/operator/mon/spec_test.go b/pkg/operator/mon/spec_test.go index 5341cc8e2..05e0fc4c4 100644 --- a/pkg/operator/mon/spec_test.go +++ b/pkg/operator/mon/spec_test.go @@ -34,7 +34,7 @@ func TestPodSpecs(t *testing.T) { func testPodSpec(t *testing.T, dataDir string) { clientset := testop.New(1) - c := New(&clusterd.Context{Clientset: clientset}, "ns", dataDir, "myversion", k8sutil.Placement{}) + c := New(&clusterd.Context{Clientset: clientset}, "ns", dataDir, "myversion", k8sutil.Placement{}, false) c.clusterInfo = testop.CreateClusterInfo(0) config := &monConfig{Name: "mon0", Port: 6790} diff --git a/pkg/operator/osd/osd.go b/pkg/operator/osd/osd.go index c41e5b653..2442e3969 100644 --- a/pkg/operator/osd/osd.go +++ b/pkg/operator/osd/osd.go @@ -51,10 +51,11 @@ type Cluster struct { Version string Storage StorageSpec dataDirHostPath string + HostNetwork bool } // New creates an instance of the OSD manager -func New(context *clusterd.Context, namespace, version string, storageSpec StorageSpec, dataDirHostPath string, placement k8sutil.Placement) *Cluster { +func New(context *clusterd.Context, namespace, version string, storageSpec StorageSpec, dataDirHostPath string, placement k8sutil.Placement, hostNetwork bool) *Cluster { return &Cluster{ context: context, Namespace: namespace, @@ -62,6 +63,7 @@ func New(context *clusterd.Context, namespace, version string, storageSpec Stora Version: version, Storage: storageSpec, dataDirHostPath: dataDirHostPath, + HostNetwork: hostNetwork, } } @@ -162,6 +164,10 @@ func (c *Cluster) podTemplateSpec(devices []Device, directories []Directory, sel Containers: []v1.Container{c.osdContainer(devices, directories, selection, config)}, RestartPolicy: v1.RestartPolicyAlways, Volumes: volumes, + HostNetwork: c.HostNetwork, + } + if c.HostNetwork { + podSpec.DNSPolicy = v1.DNSClusterFirstWithHostNet } c.placement.ApplyToPodSpec(&podSpec) diff --git a/pkg/operator/osd/osd_test.go b/pkg/operator/osd/osd_test.go index a9c8209e8..bc63f5630 100644 --- a/pkg/operator/osd/osd_test.go +++ b/pkg/operator/osd/osd_test.go @@ -30,7 +30,7 @@ import ( func TestStartDaemonset(t *testing.T) { clientset := fake.NewSimpleClientset() - c := New(&clusterd.Context{Clientset: clientset}, "ns", "myversion", StorageSpec{}, "", k8sutil.Placement{}) + c := New(&clusterd.Context{Clientset: clientset}, "ns", "myversion", StorageSpec{}, "", k8sutil.Placement{}, false) // Start the first time err := c.Start() @@ -65,7 +65,7 @@ func testPodDevices(t *testing.T, dataDir, deviceFilter string, allDevices bool) } clientset := fake.NewSimpleClientset() - c := New(&clusterd.Context{Clientset: clientset}, "ns", "myversion", storageSpec, dataDir, k8sutil.Placement{}) + c := New(&clusterd.Context{Clientset: clientset}, "ns", "myversion", storageSpec, dataDir, k8sutil.Placement{}, false) n := c.Storage.resolveNode(storageSpec.Nodes[0].Name) replicaSet := c.makeReplicaSet(n.Name, n.Devices, n.Directories, n.Selection, n.Config) @@ -138,7 +138,7 @@ func TestStorageSpecDevicesAndDirectories(t *testing.T) { } clientset := fake.NewSimpleClientset() - c := New(&clusterd.Context{Clientset: clientset}, "ns", "myversion", storageSpec, "", k8sutil.Placement{}) + c := New(&clusterd.Context{Clientset: clientset}, "ns", "myversion", storageSpec, "", k8sutil.Placement{}, false) n := c.Storage.resolveNode(storageSpec.Nodes[0].Name) replicaSet := c.makeReplicaSet(n.Name, n.Devices, n.Directories, n.Selection, n.Config) @@ -180,7 +180,7 @@ func TestStorageSpecConfig(t *testing.T) { } clientset := fake.NewSimpleClientset() - c := New(&clusterd.Context{Clientset: clientset}, "ns", "myversion", storageSpec, "", k8sutil.Placement{}) + c := New(&clusterd.Context{Clientset: clientset}, "ns", "myversion", storageSpec, "", k8sutil.Placement{}, false) n := c.Storage.resolveNode(storageSpec.Nodes[0].Name) replicaSet := c.makeReplicaSet(n.Name, n.Devices, n.Directories, n.Selection, n.Config) @@ -194,3 +194,33 @@ func TestStorageSpecConfig(t *testing.T) { verifyEnvVar(t, container.Env, "ROOK_OSD_JOURNAL_SIZE", strconv.Itoa(30), true) verifyEnvVar(t, container.Env, "ROOK_LOCATION", "rack=foo", true) } + +func TestHostNetwork(t *testing.T) { + storageSpec := StorageSpec{ + Config: Config{}, + Nodes: []Node{ + { + Name: "node1", + Config: Config{ + Location: "rack=foo", + StoreConfig: cephosd.StoreConfig{ + StoreType: cephosd.Bluestore, + DatabaseSizeMB: 10, + WalSizeMB: 20, + JournalSizeMB: 30, + }, + }, + }, + }, + } + + clientset := fake.NewSimpleClientset() + c := New(&clusterd.Context{Clientset: clientset}, "ns", "myversion", storageSpec, "", k8sutil.Placement{}, true) + + n := c.Storage.resolveNode(storageSpec.Nodes[0].Name) + r := c.makeReplicaSet(n.Name, n.Devices, n.Directories, n.Selection, n.Config) + assert.NotNil(t, r) + + assert.Equal(t, true, r.Spec.Template.Spec.HostNetwork) + assert.Equal(t, v1.DNSClusterFirstWithHostNet, r.Spec.Template.Spec.DNSPolicy) +} diff --git a/pkg/operator/rgw/rgw.go b/pkg/operator/rgw/rgw.go index e6c01727b..e84078a4b 100644 --- a/pkg/operator/rgw/rgw.go +++ b/pkg/operator/rgw/rgw.go @@ -51,11 +51,12 @@ const ( // Cluster for rgw management type Cluster struct { - context *clusterd.Context - Namespace string - placement k8sutil.Placement - Version string - Config model.ObjectStore + context *clusterd.Context + Namespace string + placement k8sutil.Placement + Version string + Config model.ObjectStore + HostNetwork bool } type idType struct { @@ -67,14 +68,15 @@ type realmType struct { } // New creates an instance of an rgw manager -func New(context *clusterd.Context, config model.ObjectStore, namespace, version string, placement k8sutil.Placement) *Cluster { +func New(context *clusterd.Context, config model.ObjectStore, namespace, version string, placement k8sutil.Placement, hostNetwork bool) *Cluster { return &Cluster{ - context: context, - Namespace: namespace, - placement: placement, - Version: version, - Config: config, + context: context, + Namespace: namespace, + placement: placement, + Version: version, + Config: config, + HostNetwork: hostNetwork, } } @@ -331,7 +333,12 @@ func (c *Cluster) makeDeployment() *extensions.Deployment { {Name: k8sutil.DataDirVolume, VolumeSource: v1.VolumeSource{EmptyDir: &v1.EmptyDirVolumeSource{}}}, k8sutil.ConfigOverrideVolume(), }, + HostNetwork: c.HostNetwork, } + if c.HostNetwork { + podSpec.DNSPolicy = v1.DNSClusterFirstWithHostNet + } + c.placement.ApplyToPodSpec(&podSpec) // Set the ssl cert if specified if c.Config.CertificateRef != "" { @@ -419,6 +426,9 @@ func (c *Cluster) startService() (string, error) { Selector: labels, }, } + if c.HostNetwork { + s.Spec.ClusterIP = v1.ClusterIPNone + } s, err := c.context.Clientset.CoreV1().Services(c.Namespace).Create(s) if err != nil { diff --git a/pkg/operator/rgw/rgw_test.go b/pkg/operator/rgw/rgw_test.go index 054e4b26b..b34bd8fc0 100644 --- a/pkg/operator/rgw/rgw_test.go +++ b/pkg/operator/rgw/rgw_test.go @@ -47,7 +47,7 @@ func TestStartRGW(t *testing.T) { defer os.RemoveAll(configDir) config := model.ObjectStore{Name: "default", Port: 123} context := &clusterd.Context{Clientset: clientset, Executor: executor, ConfigDir: configDir} - c := New(context, config, "ns", "version", k8sutil.Placement{}) + c := New(context, config, "ns", "version", k8sutil.Placement{}, false) // start a basic cluster err := c.Start() @@ -81,7 +81,7 @@ func validateStart(t *testing.T, c *Cluster, clientset *fake.Clientset) { func TestPodSpecs(t *testing.T) { config := model.ObjectStore{Name: "default", Port: 123} - c := New(nil, config, "ns", "myversion", k8sutil.Placement{}) + c := New(nil, config, "ns", "myversion", k8sutil.Placement{}, false) d := c.makeDeployment() assert.NotNil(t, d) @@ -111,7 +111,7 @@ func TestPodSpecs(t *testing.T) { func TestSSLPodSpec(t *testing.T) { config := model.ObjectStore{Name: "default", Port: 123, CertificateRef: "mycert"} - c := New(nil, config, "ns", "myversion", k8sutil.Placement{}) + c := New(nil, config, "ns", "myversion", k8sutil.Placement{}, false) d := c.makeDeployment() assert.NotNil(t, d) @@ -147,9 +147,8 @@ func TestCreateRealm(t *testing.T) { } else if args[0] == "realm" && args[1] == "list" { if defaultStore { return "", fmt.Errorf("failed to run radosgw-admin: Failed to complete : exit status 2") - } else { - return `{"realms": ["myobj"]}`, nil } + return `{"realms": ["myobj"]}`, nil } return idResponse, nil }, @@ -157,7 +156,7 @@ func TestCreateRealm(t *testing.T) { config := model.ObjectStore{Name: "myobject", Port: 123} context := &clusterd.Context{Executor: executor} - c := New(context, config, "ns", "version", k8sutil.Placement{}) + c := New(context, config, "ns", "version", k8sutil.Placement{}, false) // create the first realm, marked as default err := c.createRealm("1.2.3.4") @@ -168,3 +167,14 @@ func TestCreateRealm(t *testing.T) { err = c.createRealm("2.3.4.5") assert.Nil(t, err) } + +func TestHostNetwork(t *testing.T) { + config := model.ObjectStore{Name: "default", Port: 123, CertificateRef: "mycert"} + c := New(nil, config, "ns", "myversion", k8sutil.Placement{}, true) + + d := c.makeDeployment() + assert.NotNil(t, d) + + assert.Equal(t, true, d.Spec.Template.Spec.HostNetwork) + assert.Equal(t, v1.DNSClusterFirstWithHostNet, d.Spec.Template.Spec.DNSPolicy) +} diff --git a/pkg/operator/test/client.go b/pkg/operator/test/client.go index 7c31a785d..24b5753cf 100644 --- a/pkg/operator/test/client.go +++ b/pkg/operator/test/client.go @@ -29,7 +29,19 @@ func New(nodes int) *fake.Clientset { clientset := fake.NewSimpleClientset() for i := 0; i < nodes; i++ { ready := v1.NodeCondition{Type: v1.NodeReady} - n := &v1.Node{Status: v1.NodeStatus{Conditions: []v1.NodeCondition{ready}}} + n := &v1.Node{ + Status: v1.NodeStatus{ + Conditions: []v1.NodeCondition{ + ready, + }, + Addresses: []v1.NodeAddress{ + { + Type: v1.NodeExternalIP, + Address: fmt.Sprintf("%d.%d.%d.%d", i, i, i, i), + }, + }, + }, + } n.Name = fmt.Sprintf("node%d", i) clientset.CoreV1().Nodes().Create(n) } diff --git a/tests/framework/installer/install_data.go b/tests/framework/installer/install_data.go index f2e4eebc5..ed0354f75 100644 --- a/tests/framework/installer/install_data.go +++ b/tests/framework/installer/install_data.go @@ -48,7 +48,7 @@ spec: args: ["operator", "--mon-healthcheck-interval=5s", "--mon-out-timeout=1s"] env: - name: ROOK_REPO_PREFIX - value: roo` + value: rook` } return `kind: ClusterRole @@ -167,7 +167,9 @@ spec: args: ["operator", "--mon-healthcheck-interval=5s", "--mon-out-timeout=1s"] env: - name: ROOK_REPO_PREFIX - value: rook` + value: rook + - name: ROOK_LOG_LEVEL + value: DEBUG` } @@ -186,6 +188,7 @@ metadata: spec: versionTag: master dataDirHostPath: + hostNetwork: false # To control where various services will be scheduled by kubernetes, use the placement configuration sections below. # The example under 'all' would have all services scheduled on kubernetes nodes labeled with 'role=storage' and # tolerate taints with a key of 'storage-node'.