ceph: mon service forwards default bind port 6789

The Ceph mon flag `--public-bind-addr` does not keep the same port as
the previous daemon run when the port is left off the flag's value. A
port is undesired since that could interfere with Nautilus' msgr2
protocol, so instead configure the mon's services to forward the default
bind port on pods (6789) to the mon endpoint's port. The endpoint port
will be 6789 for new mons, but it may be 6790 for legacy clusters.

Signed-off-by: Blaine Gardner <blaine.gardner@suse.com>

pkg/operator/ceph/cluster/mon/health.go

@@ -324,7 +324,7 @@ func (c *Cluster) removeMon(daemonName string) error {
 		delete(c.mapping.Node, daemonName)
 		// if node->port "mapping" has been created, decrease or delete it
 		if port, ok := c.mapping.Port[nodeName]; ok {
-			if port == DefaultPort {
+			if port == DefaultMsgr1Port {
 				delete(c.mapping.Port, nodeName)
 			}
 			// don't clean up if a node port is higher than the default port, other

pkg/operator/ceph/cluster/mon/health_test.go

@@ -38,6 +38,7 @@ import (
 )
 
 func TestCheckHealth(t *testing.T) {
+
 	var deploymentsUpdated *[]*apps.Deployment
 	updateDeploymentAndWait, deploymentsUpdated = testopk8s.UpdateDeploymentAndWaitStub()
 
@@ -64,7 +65,7 @@ func TestCheckHealth(t *testing.T) {
 		Name:    "node0",
 		Address: "",
 	}
-	c.mapping.Port["node0"] = DefaultPort
+	c.mapping.Port["node0"] = DefaultMsgr1Port
 	c.maxMonID = 4
 
 	err := c.checkHealth()
@@ -117,7 +118,7 @@ func TestCheckHealthNotFound(t *testing.T) {
 	c.mapping.Node["b"] = &NodeInfo{
 		Name: "node0",
 	}
-	c.mapping.Port["node0"] = DefaultPort
+	c.mapping.Port["node0"] = DefaultMsgr1Port
 	c.maxMonID = 4
 
 	c.saveMonConfig()

pkg/operator/ceph/cluster/mon/mon.go

@@ -66,16 +66,20 @@ const (
 	DefaultMonCount = 3
 	// MaxMonCount Maximum allowed mon count for a cluster
 	MaxMonCount = 9
-	// Msgr2port is the listening port of the messenger v2 protocol
-	Msgr2port int32 = 3300
+
+	// DefaultMsgr1Port is the default port Ceph mons use to communicate amongst themselves prior
+	// to Ceph Nautilus.
+	DefaultMsgr1Port int32 = 6789
+	// DefaultMsgr2Port is the listening port of the messenger v2 protocol introduced in Ceph
+	// Nautilus. In Nautilus and a few Ceph releases after, Ceph can use both v1 and v2 protocols.
+	DefaultMsgr2Port int32 = 3300
+
 	// minimum amount of memory in MB to run the pod
 	cephMonPodMinimumMemory uint64 = 1024
 )
 
 var (
 	logger = capnslog.NewPackageLogger("github.com/rook/rook", "op-mon")
-	// DefaultPort is the default port Ceph mons use to communicate amongst themselves.
-	DefaultPort int32 = 6789
 )
 
 // Cluster represents the Rook and environment configuration settings needed to set up Ceph mons.
@@ -317,7 +321,7 @@ func (c *Cluster) newMonConfig(monID int) *monConfig {
 	return &monConfig{
 		ResourceName: resourceName(daemonName),
 		DaemonName:   daemonName,
-		Port:         DefaultPort,
+		Port:         DefaultMsgr1Port,
 		DataPathMap: config.NewStatefulDaemonDataPathMap(
 			c.dataDirHostPath, dataDirRelativeHostPath(daemonName), config.MonType, daemonName),
 	}

pkg/operator/ceph/cluster/mon/mon_test.go

@@ -62,7 +62,7 @@ func testGenMonConfig(monID string) *monConfig {
 	return &monConfig{
 		ResourceName: "rook-ceph-" + moniker, // rook-ceph-mon-A or rook-ceph-mon#
 		DaemonName:   monID,                  // A or mon#
-		Port:         DefaultPort,
+		Port:         DefaultMsgr1Port,
 		PublicIP:     fmt.Sprintf("2.4.6.%d", index+1),
 		// dataDirHostPath assumed to be /var/lib/rook
 		DataPathMap: config.NewStatefulDaemonDataPathMap(
@@ -142,6 +142,7 @@ func TestResourceName(t *testing.T) {
 }
 
 func TestStartMonPods(t *testing.T) {
+
 	namespace := "ns"
 	context := newTestStartCluster(namespace)
 	c := newCluster(context, namespace, false, true, v1.ResourceRequirements{})
@@ -160,6 +161,7 @@ func TestStartMonPods(t *testing.T) {
 }
 
 func TestOperatorRestart(t *testing.T) {
+
 	namespace := "ns"
 	context := newTestStartCluster(namespace)
 	c := newCluster(context, namespace, false, true, v1.ResourceRequirements{})
@@ -184,6 +186,7 @@ func TestOperatorRestart(t *testing.T) {
 
 // safety check that if hostNetwork is used no changes occur on an operator restart
 func TestOperatorRestartHostNetwork(t *testing.T) {
+
 	namespace := "ns"
 	context := newTestStartCluster(namespace)
 

pkg/operator/ceph/cluster/mon/service.go

@@ -22,14 +22,13 @@ import (
 
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	v1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 )
 
 func (c *Cluster) createService(mon *monConfig) (string, error) {
 	labels := c.getLabels(mon.DaemonName)
-	s := &v1.Service{
+	svcDef := &v1.Service{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:   mon.ResourceName,
 			Labels: labels,
@@ -37,38 +36,34 @@ func (c *Cluster) createService(mon *monConfig) (string, error) {
 		Spec: v1.ServiceSpec{
 			Ports: []v1.ServicePort{
 				{
-					Name:       mon.ResourceName,
-					Port:       mon.Port,
-					TargetPort: intstr.FromInt(int(mon.Port)),
+					Name: "msgr1",
+					Port: mon.Port,
+					// --public-bind-addr=IP with no IP:port has the mon listen on port 6789
+					// regardless of what port the mon advertises (--public-addr) to the outside.
+					TargetPort: intstr.FromInt(int(DefaultMsgr1Port)),
 					Protocol:   v1.ProtocolTCP,
 				},
 			},
 			Selector: labels,
 		},
 	}
-	k8sutil.SetOwnerRef(c.context.Clientset, c.Namespace, &s.ObjectMeta, &c.ownerRef)
+	k8sutil.SetOwnerRef(c.context.Clientset, c.Namespace, &svcDef.ObjectMeta, &c.ownerRef)
 	if c.HostNetwork {
-		s.Spec.ClusterIP = v1.ClusterIPNone
+		svcDef.Spec.ClusterIP = v1.ClusterIPNone
 	}
 
 	// If deploying Nautilus or newer we need a new port for the monitor service
 	if c.clusterInfo.CephVersion.IsAtLeastNautilus() {
-		addServicePort(s, "msgr2", Msgr2port)
+		addServicePort(svcDef, "msgr2", DefaultMsgr2Port)
 	}
 
-	s, err := c.context.Clientset.CoreV1().Services(c.Namespace).Create(s)
+	s, err := k8sutil.CreateOrUpdateService(c.context.Clientset, c.Namespace, svcDef)
 	if err != nil {
-		if !errors.IsAlreadyExists(err) {
-			return "", fmt.Errorf("failed to create mon service. %+v", err)
-		}
-		s, err = c.context.Clientset.CoreV1().Services(c.Namespace).Get(mon.ResourceName, metav1.GetOptions{})
-		if err != nil {
-			return "", fmt.Errorf("failed to get mon %s service ip. %+v", mon.ResourceName, err)
-		}
+		return "", fmt.Errorf("failed to create service for mon %s. %+v", mon.DaemonName, err)
 	}
 
 	if s == nil {
-		logger.Warningf("service ip not found for mon %s. this better be a test", mon.ResourceName)
+		logger.Errorf("service ip not found for mon %s. if this is not a unit test, this is an error", mon.ResourceName)
 		return "", nil
 	}
 
@@ -76,7 +71,7 @@ func (c *Cluster) createService(mon *monConfig) (string, error) {
 	// however it's interesting to show that monitors can be addressed via 2 different ports
 	// in the end the service has msgr1 and msgr2 ports configured so it's not entirely wrong
 	if c.clusterInfo.CephVersion.IsAtLeastNautilus() {
-		logger.Infof("mon %s endpoint are [v2:%s:%s,v1:%s:%d]", mon.DaemonName, s.Spec.ClusterIP, strconv.Itoa(int(Msgr2port)), s.Spec.ClusterIP, mon.Port)
+		logger.Infof("mon %s endpoint are [v2:%s:%s,v1:%s:%d]", mon.DaemonName, s.Spec.ClusterIP, strconv.Itoa(int(DefaultMsgr2Port)), s.Spec.ClusterIP, mon.Port)
 	} else {
 		logger.Infof("mon %s endpoint is %s:%d", mon.DaemonName, s.Spec.ClusterIP, mon.Port)
 	}

pkg/operator/ceph/cluster/mon/spec.go

@@ -164,7 +164,11 @@ func (c *Cluster) makeMonDaemonContainer(monConfig *monConfig) v1.Container {
 		Args: append(
 			opspec.DaemonFlags(c.clusterInfo, monConfig.DaemonName),
 			"--foreground",
+			// If the mon is already in the monmap, when the port is left off of --public-addr,
+			// it will still advertise on the previous port b/c monmap is saved to mon database.
 			config.NewFlag("public-addr", monConfig.PublicIP),
+			// Opposite of the above, --public-bind-addr will *not* still advertise on the previous
+			// port, which makes sense because this is the pod IP, which changes with every new pod.
 			config.NewFlag("public-bind-addr", opspec.ContainerEnvVarReference(podIPEnvVar)),
 		),
 		Image:           c.spec.CephVersion.Image,

pkg/operator/k8sutil/daemonset.go

@@ -0,0 +1,52 @@
+/*
+Copyright 2018 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package k8sutil
+
+import (
+	"fmt"
+
+	apps "k8s.io/api/apps/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+// CreateDaemonSet creates
+func CreateDaemonSet(name, namespace string, clientset kubernetes.Interface, ds *apps.DaemonSet) error {
+	_, err := clientset.Apps().DaemonSets(namespace).Create(ds)
+	if err != nil {
+		if k8serrors.IsAlreadyExists(err) {
+			_, err = clientset.Apps().DaemonSets(namespace).Update(ds)
+		}
+		if err != nil {
+			return fmt.Errorf("failed to start %s daemonset: %v\n%v", name, err, ds)
+		}
+	}
+	return err
+}
+
+// DeleteDaemonset makes a best effort at deleting a daemonset and its pods, then waits for them to be deleted
+func DeleteDaemonset(clientset kubernetes.Interface, namespace, name string) error {
+	deleteAction := func(options *metav1.DeleteOptions) error {
+		return clientset.Apps().DaemonSets(namespace).Delete(name, options)
+	}
+	getAction := func() error {
+		_, err := clientset.Apps().DaemonSets(namespace).Get(name, metav1.GetOptions{})
+		return err
+	}
+	return deleteResourceAndWait(namespace, name, "daemonset", deleteAction, getAction)
+}

pkg/operator/k8sutil/deployment.go

@@ -13,6 +13,7 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
 */
+
 package k8sutil
 
 import (
@@ -130,3 +131,16 @@ func GetDeployments(clientset kubernetes.Interface, namespace, labelSelector str
 	}
 	return deployments, nil
 }
+
+// DeleteDeployment makes a best effort at deleting a deployment and its pods, then waits for them to be deleted
+func DeleteDeployment(clientset kubernetes.Interface, namespace, name string) error {
+	logger.Debugf("removing %s deployment if it exists", name)
+	deleteAction := func(options *metav1.DeleteOptions) error {
+		return clientset.Apps().Deployments(namespace).Delete(name, options)
+	}
+	getAction := func() error {
+		_, err := clientset.Apps().Deployments(namespace).Get(name, metav1.GetOptions{})
+		return err
+	}
+	return deleteResourceAndWait(namespace, name, "deployment", deleteAction, getAction)
+}

pkg/operator/k8sutil/k8sutil.go

@@ -21,8 +21,11 @@ import (
 	"crypto/md5"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/coreos/pkg/capnslog"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/kubernetes/pkg/util/version"
@@ -85,3 +88,43 @@ func TruncateNodeName(format, nodeName string) string {
 	}
 	return fmt.Sprintf(format, nodeName)
 }
+
+// deleteResourceAndWait will delete a resource, then wait for it to be purged from the system
+func deleteResourceAndWait(namespace, name, resourceType string,
+	deleteAction func(*metav1.DeleteOptions) error,
+	getAction func() error,
+) error {
+	var gracePeriod int64
+	propagation := metav1.DeletePropagationForeground
+	options := &metav1.DeleteOptions{GracePeriodSeconds: &gracePeriod, PropagationPolicy: &propagation}
+
+	// Delete the resource if it exists
+	logger.Infof("removing %s %s if it exists", resourceType, name)
+	err := deleteAction(options)
+	if err != nil {
+		if !errors.IsNotFound(err) {
+			return fmt.Errorf("failed to delete %s. %+v", name, err)
+		}
+		return nil
+	}
+	logger.Infof("Removed %s %s", resourceType, name)
+
+	// wait for the resource to be deleted
+	sleepTime := 2 * time.Second
+	for i := 0; i < 30; i++ {
+		// check for the existence of the resource
+		err = getAction()
+		if err != nil {
+			if errors.IsNotFound(err) {
+				logger.Infof("confirmed %s does not exist", name)
+				return nil
+			}
+			return fmt.Errorf("failed to get %s. %+v", name, err)
+		}
+
+		logger.Infof("%s still found. waiting...", name)
+		time.Sleep(sleepTime)
+	}
+
+	return fmt.Errorf("gave up waiting for %s pods to be terminated", name)
+}

pkg/operator/k8sutil/name.go

@@ -10,6 +10,7 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
 */
+
 package k8sutil
 
 import "fmt"

pkg/operator/k8sutil/pod.go

@@ -23,13 +23,11 @@ import (
 	"os"
 	"path"
 	"strings"
-	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 
 	v1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 )
 
@@ -220,83 +218,6 @@ func GetPodLog(clientset kubernetes.Interface, namespace string, labelSelector s
 	return "", fmt.Errorf("did not find any pods with label %s", labelSelector)
 }
 
-// DeleteDeployment makes a best effort at deleting a deployment and its pods, then waits for them to be deleted
-func DeleteDeployment(clientset kubernetes.Interface, namespace, name string) error {
-	logger.Debugf("removing %s deployment if it exists", name)
-	deleteAction := func(options *metav1.DeleteOptions) error {
-		return clientset.Apps().Deployments(namespace).Delete(name, options)
-	}
-	getAction := func() error {
-		_, err := clientset.Apps().Deployments(namespace).Get(name, metav1.GetOptions{})
-		return err
-	}
-	return deleteResourceAndWait(namespace, name, "deployment", deleteAction, getAction)
-}
-
-// DeleteReplicaSet makes a best effort at deleting a deployment and its pods, then waits for them to be deleted
-func DeleteReplicaSet(clientset kubernetes.Interface, namespace, name string) error {
-	deleteAction := func(options *metav1.DeleteOptions) error {
-		return clientset.Apps().ReplicaSets(namespace).Delete(name, options)
-	}
-	getAction := func() error {
-		_, err := clientset.Apps().ReplicaSets(namespace).Get(name, metav1.GetOptions{})
-		return err
-	}
-	return deleteResourceAndWait(namespace, name, "replicaset", deleteAction, getAction)
-}
-
-// DeleteDaemonset makes a best effort at deleting a daemonset and its pods, then waits for them to be deleted
-func DeleteDaemonset(clientset kubernetes.Interface, namespace, name string) error {
-	logger.Infof("removing %s daemonset if it exists", name)
-	deleteAction := func(options *metav1.DeleteOptions) error {
-		return clientset.Apps().DaemonSets(namespace).Delete(name, options)
-	}
-	getAction := func() error {
-		_, err := clientset.Apps().DaemonSets(namespace).Get(name, metav1.GetOptions{})
-		return err
-	}
-	return deleteResourceAndWait(namespace, name, "daemonset", deleteAction, getAction)
-}
-
-// deleteResourceAndWait will delete a resource, then wait for it to be purged from the system
-func deleteResourceAndWait(namespace, name, resourceType string,
-	deleteAction func(*metav1.DeleteOptions) error,
-	getAction func() error) error {
-
-	var gracePeriod int64
-	propagation := metav1.DeletePropagationForeground
-	options := &metav1.DeleteOptions{GracePeriodSeconds: &gracePeriod, PropagationPolicy: &propagation}
-
-	// Delete the resource if it exists
-	err := deleteAction(options)
-	if err != nil {
-		if !errors.IsNotFound(err) {
-			return fmt.Errorf("failed to delete %s. %+v", name, err)
-		}
-		return nil
-	}
-	logger.Infof("Removed %s %s", resourceType, name)
-
-	// wait for the resource to be deleted
-	sleepTime := 2 * time.Second
-	for i := 0; i < 30; i++ {
-		// check for the existence of the resource
-		err = getAction()
-		if err != nil {
-			if errors.IsNotFound(err) {
-				logger.Infof("confirmed %s does not exist", name)
-				return nil
-			}
-			return fmt.Errorf("failed to get %s. %+v", name, err)
-		}
-
-		logger.Infof("%s still found. waiting...", name)
-		time.Sleep(sleepTime)
-	}
-
-	return fmt.Errorf("gave up waiting for %s pods to be terminated", name)
-}
-
 // ClusterDaemonEnvVars Environment variables used by storage cluster daemon
 func ClusterDaemonEnvVars(image string) []v1.EnvVar {
 	return []v1.EnvVar{
@@ -322,8 +243,3 @@ func ClusterDaemonEnvVars(image string) []v1.EnvVar {
 		{Name: "POD_CPU_REQUEST", ValueFrom: &v1.EnvVarSource{ResourceFieldRef: &v1.ResourceFieldSelector{Resource: "requests.cpu"}}},
 	}
 }
-
-// CephVersionName is the name of the ceph version
-func CephVersionName(version string) v1.EnvVar {
-	return v1.EnvVar{Name: "ROOK_CEPH_VERSION_NAME", Value: version}
-}

pkg/operator/k8sutil/replicaset.go

@@ -0,0 +1,34 @@
+/*
+Copyright 2018 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package k8sutil
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+// DeleteReplicaSet makes a best effort at deleting a deployment and its pods, then waits for them to be deleted
+func DeleteReplicaSet(clientset kubernetes.Interface, namespace, name string) error {
+	deleteAction := func(options *metav1.DeleteOptions) error {
+		return clientset.Apps().ReplicaSets(namespace).Delete(name, options)
+	}
+	getAction := func() error {
+		_, err := clientset.Apps().ReplicaSets(namespace).Get(name, metav1.GetOptions{})
+		return err
+	}
+	return deleteResourceAndWait(namespace, name, "replicaset", deleteAction, getAction)
+}

pkg/operator/k8sutil/service.go

@@ -0,0 +1,63 @@
+/*
+Copyright 2019 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package k8sutil
+
+import (
+	"fmt"
+
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+// CreateOrUpdateService creates a service or updates the service declaratively if it already exists.
+func CreateOrUpdateService(
+	clientset kubernetes.Interface, namespace string, serviceDefinition *v1.Service,
+) (*v1.Service, error) {
+	name := serviceDefinition.Name
+	logger.Debugf("creating service %s", name)
+	s, err := clientset.CoreV1().Services(namespace).Create(serviceDefinition)
+	if err != nil {
+		if !errors.IsAlreadyExists(err) {
+			return nil, fmt.Errorf("failed to create service %s. %+v", name, err)
+		}
+		s, err = UpdateService(clientset, namespace, serviceDefinition)
+		if err != nil {
+			return nil, fmt.Errorf("failed to update service %s. %+v", name, err)
+		}
+	}
+	return s, err
+}
+
+// UpdateService updates a service declaratively. If the service does not exist this is considered
+// an error condition.
+func UpdateService(
+	clientset kubernetes.Interface, namespace string, serviceDefinition *v1.Service,
+) (*v1.Service, error) {
+	name := serviceDefinition.Name
+	logger.Debug("updating service %s")
+	existing, err := clientset.CoreV1().Services(namespace).Get(name, metav1.GetOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("could not get existing service %s in order to update. %+v", name, err)
+	}
+	// ClusterIP is immutable for k8s services and cannot be left empty in k8s v1 API
+	serviceDefinition.Spec.ClusterIP = existing.Spec.ClusterIP
+	// ResourceVersion required to update services in k8s v1 API to prevent race conditions
+	serviceDefinition.ResourceVersion = existing.ResourceVersion
+	return clientset.CoreV1().Services(namespace).Update(serviceDefinition)
+}

pkg/operator/k8sutil/statefulset.go

@@ -47,20 +47,6 @@ func makeHeadlessSvc(name, namespace string, label map[string]string, clientset
 	return svc, nil
 }
 
-// create a apps.daemonset
-func CreateDaemonSet(name, namespace string, clientset kubernetes.Interface, ds *apps.DaemonSet) error {
-	_, err := clientset.Apps().DaemonSets(namespace).Create(ds)
-	if err != nil {
-		if k8serrors.IsAlreadyExists(err) {
-			_, err = clientset.Apps().DaemonSets(namespace).Update(ds)
-		}
-		if err != nil {
-			return fmt.Errorf("failed to start %s daemonset: %v\n%v", name, err, ds)
-		}
-	}
-	return err
-}
-
 // create a apps.statefulset and a headless svc
 func CreateStatefulSet(name, namespace, appName string, clientset kubernetes.Interface, ss *apps.StatefulSet) (*corev1.Service, error) {
 	label := ss.GetLabels()