From feff587992336ca6c4f2cc48e59aec6f04f4977e Mon Sep 17 00:00:00 2001 From: Blaine Gardner Date: Mon, 1 Apr 2019 16:20:35 -0600 Subject: [PATCH] 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 --- pkg/operator/ceph/cluster/mon/health.go | 2 +- pkg/operator/ceph/cluster/mon/health_test.go | 5 +- pkg/operator/ceph/cluster/mon/mon.go | 14 ++-- pkg/operator/ceph/cluster/mon/mon_test.go | 5 +- pkg/operator/ceph/cluster/mon/service.go | 31 +++---- pkg/operator/ceph/cluster/mon/spec.go | 4 + pkg/operator/k8sutil/daemonset.go | 52 ++++++++++++ pkg/operator/k8sutil/deployment.go | 14 ++++ pkg/operator/k8sutil/k8sutil.go | 43 ++++++++++ pkg/operator/k8sutil/name.go | 1 + pkg/operator/k8sutil/pod.go | 84 ------------------- pkg/operator/k8sutil/replicaset.go | 34 ++++++++ pkg/operator/k8sutil/service.go | 63 ++++++++++++++ .../k8sutil/{apps.go => statefulset.go} | 14 ---- 14 files changed, 241 insertions(+), 125 deletions(-) create mode 100644 pkg/operator/k8sutil/daemonset.go create mode 100644 pkg/operator/k8sutil/replicaset.go create mode 100644 pkg/operator/k8sutil/service.go rename pkg/operator/k8sutil/{apps.go => statefulset.go} (83%) 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()