diff --git a/pkg/operator/ceph/cluster/mon/health.go b/pkg/operator/ceph/cluster/mon/health.go index 156bf8751..f79cbd5b1 100644 --- a/pkg/operator/ceph/cluster/mon/health.go +++ b/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 diff --git a/pkg/operator/ceph/cluster/mon/health_test.go b/pkg/operator/ceph/cluster/mon/health_test.go index d101323ea..b43e58d2b 100644 --- a/pkg/operator/ceph/cluster/mon/health_test.go +++ b/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() diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go index a051495cd..f94db27a7 100644 --- a/pkg/operator/ceph/cluster/mon/mon.go +++ b/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), } diff --git a/pkg/operator/ceph/cluster/mon/mon_test.go b/pkg/operator/ceph/cluster/mon/mon_test.go index 47135e00a..0fdf11299 100644 --- a/pkg/operator/ceph/cluster/mon/mon_test.go +++ b/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) diff --git a/pkg/operator/ceph/cluster/mon/service.go b/pkg/operator/ceph/cluster/mon/service.go index 8218d92cd..6e9c585b8 100644 --- a/pkg/operator/ceph/cluster/mon/service.go +++ b/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) } diff --git a/pkg/operator/ceph/cluster/mon/spec.go b/pkg/operator/ceph/cluster/mon/spec.go index 5c3b58f3c..0f6e91bbf 100644 --- a/pkg/operator/ceph/cluster/mon/spec.go +++ b/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, diff --git a/pkg/operator/k8sutil/daemonset.go b/pkg/operator/k8sutil/daemonset.go new file mode 100644 index 000000000..41ea47921 --- /dev/null +++ b/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) +} diff --git a/pkg/operator/k8sutil/deployment.go b/pkg/operator/k8sutil/deployment.go index 06f2ac4f7..8acef5a25 100644 --- a/pkg/operator/k8sutil/deployment.go +++ b/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) +} diff --git a/pkg/operator/k8sutil/k8sutil.go b/pkg/operator/k8sutil/k8sutil.go index 3eae4814b..c09271d1d 100644 --- a/pkg/operator/k8sutil/k8sutil.go +++ b/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) +} diff --git a/pkg/operator/k8sutil/name.go b/pkg/operator/k8sutil/name.go index d2129e3e0..45c18c9ee 100644 --- a/pkg/operator/k8sutil/name.go +++ b/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" diff --git a/pkg/operator/k8sutil/pod.go b/pkg/operator/k8sutil/pod.go index 9ea64bdfe..366f63e6d 100644 --- a/pkg/operator/k8sutil/pod.go +++ b/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} -} diff --git a/pkg/operator/k8sutil/replicaset.go b/pkg/operator/k8sutil/replicaset.go new file mode 100644 index 000000000..58614bc97 --- /dev/null +++ b/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) +} diff --git a/pkg/operator/k8sutil/service.go b/pkg/operator/k8sutil/service.go new file mode 100644 index 000000000..159b23e5c --- /dev/null +++ b/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) +} diff --git a/pkg/operator/k8sutil/apps.go b/pkg/operator/k8sutil/statefulset.go similarity index 83% rename from pkg/operator/k8sutil/apps.go rename to pkg/operator/k8sutil/statefulset.go index 99b6ac9d9..14258b431 100644 --- a/pkg/operator/k8sutil/apps.go +++ b/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()