Merge pull request #975 from galexrt/feature/hostnetwork

Add hostNetwork option to cluster spec

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.

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'.

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
+        path: /lib/modules

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)

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)

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()
 }
 

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

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,

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 {

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++

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),

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()

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)

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.

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)
 

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)
+}

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)

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)
+}

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
 }

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
 }
 

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])
+}

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)
 

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}
 

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)
 

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)
+}

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 {

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)
+}

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)
 	}

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'.