You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@submarine.apache.org by ka...@apache.org on 2021/08/26 14:44:09 UTC
[submarine] branch master updated: SUBMARINE-950. Create state
machine for submarine custom resource
This is an automated email from the ASF dual-hosted git repository.
kaihsun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/submarine.git
The following commit(s) were added to refs/heads/master by this push:
new 314cb00 SUBMARINE-950. Create state machine for submarine custom resource
314cb00 is described below
commit 314cb000c36e21eb572b9087cbf1719533d25b68
Author: Kenchu123 <k8...@gmail.com>
AuthorDate: Thu Aug 26 17:27:40 2021 +0800
SUBMARINE-950. Create state machine for submarine custom resource
### What is this PR for?
<!-- A few sentences describing the overall goals of the pull request's commits.
First time? Check out the contributing guide - https://submarine.apache.org/contribution/contributions.html
-->
We have created the submarine custom resource and controlled it by the operator. However, we want to know the states such as "New", "Creating", "Running", and "Failed" for the submarine custom resource, so we can watch the states of it.
States:
1. NewState: initial state
2. CreatingState: after adding a submarine CR and waiting for the pods to be READY
3. RunningState: after all pods are READY
4. FailedState: when errors occur
State Machine:
```
//+-----------------------------------------------------------------+
//| +---------+ +----------+ +----------+ |
//| | | | | | | |
//| | New +---------> Creating +----------> Running | |
//| | | | | | | |
//| +----+----+ +-----+----+ +-----+----+ |
//| | | | |
//| | | | |
//| | | | |
//| | | +-----v----+ |
//| | | | | |
//| +--------------------+---------------> Failed | |
//| | | |
//| +----------+ |
//+-----------------------------------------------------------------+
```
Reference:
https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/blob/master/pkg/controller/sparkapplication/controller.go
### What type of PR is it?
[Improvement]
### Todos
* [x] - Add state machine logic in sync handler
* [x] - Add event record for any state change
### What is the Jira issue?
<!-- * Open an issue on Jira https://issues.apache.org/jira/browse/SUBMARINE/
* Put link here, and add [SUBMARINE-*Jira number*] in PR title, eg. `SUBMARINE-23. PR title`
-->
https://issues.apache.org/jira/browse/SUBMARINE-950
### How should this be tested?
<!--
* First time? Setup Travis CI as described on https://submarine.apache.org/contribution/contributions.html#continuous-integration
* Strongly recommended: add automated unit tests for any new or changed behavior
* Outline any manual steps to test the PR here.
-->
### Screenshots (if appropriate)
https://user-images.githubusercontent.com/17617373/130261443-41ca8100-b2e5-40ba-ad03-7f4b80345e62.mov
### Questions:
* Do the license files need updating? No
* Are there breaking changes for older versions? No
* Does this need new documentation? No
Author: Kenchu123 <k8...@gmail.com>
Signed-off-by: Kai-Hsun Chen <ka...@apache.org>
Closes #719 from Kenchu123/SUBMARINE-950 and squashes the following commits:
cec76d6f [Kenchu123] SUBMARINE-950. Check deployment progressing failed during creating state
ba6bded1 [Kenchu123] SUBMARINE-950. Modify code depends on reviewers' comments
877d429b [Kenchu123] SUBMARINE-950. Add Phase PodFailed when checking submarine status
bea61a69 [Kenchu123] SUBMARINE-950. Add state machine graph
1be415a7 [Kenchu123] SUBMARINE-950. Submarine will switch to RUNNING state when all dependents are ready
41235384 [Kenchu123] SUBMARINE-950. Add Submarine status state flow and event record
---
.../pkg/apis/submarine/v1alpha1/types.go | 19 ++
.../submarine/v1alpha1/zz_generated.deepcopy.go | 17 ++
submarine-cloud-v2/pkg/controller/controller.go | 275 +++++++++++++++++----
.../pkg/controller/submarine_database.go | 20 +-
.../pkg/controller/submarine_minio.go | 6 +-
.../pkg/controller/submarine_mlflow.go | 6 +-
.../pkg/controller/submarine_server.go | 28 +--
.../pkg/controller/submarine_tensorboard.go | 6 +-
8 files changed, 292 insertions(+), 85 deletions(-)
diff --git a/submarine-cloud-v2/pkg/apis/submarine/v1alpha1/types.go b/submarine-cloud-v2/pkg/apis/submarine/v1alpha1/types.go
index f44314b..3fb2830 100644
--- a/submarine-cloud-v2/pkg/apis/submarine/v1alpha1/types.go
+++ b/submarine-cloud-v2/pkg/apis/submarine/v1alpha1/types.go
@@ -77,10 +77,29 @@ type SubmarineSpec struct {
Storage *SubmarineStorage `json:"storage"`
}
+// SubmarineStateType represents the type of the current state of a submarine.
+type SubmarineStateType string
+
+// Different states a submarine may have.
+const (
+ NewState SubmarineStateType = ""
+ CreatingState SubmarineStateType = "CREATING"
+ RunningState SubmarineStateType = "RUNNING"
+ FailedState SubmarineStateType = "FAILED"
+)
+
+// SubmarineState tells the current state of the submarine and an error message in case of failures.
+type SubmarineState struct {
+ State SubmarineStateType `json:"state"`
+ ErrorMessage string `json:"errorMessage,omitempty"`
+}
+
// SubmarineStatus is the status for a Submarine resource
type SubmarineStatus struct {
AvailableServerReplicas int32 `json:"availableServerReplicas"`
AvailableDatabaseReplicas int32 `json:"availableDatabaseReplicas"`
+ // SubmarineState tells the overall submarine state.
+ SubmarineState `json:"submarineState,omitempty"`
}
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
diff --git a/submarine-cloud-v2/pkg/apis/submarine/v1alpha1/zz_generated.deepcopy.go b/submarine-cloud-v2/pkg/apis/submarine/v1alpha1/zz_generated.deepcopy.go
index 597f6c8..ee083d3 100644
--- a/submarine-cloud-v2/pkg/apis/submarine/v1alpha1/zz_generated.deepcopy.go
+++ b/submarine-cloud-v2/pkg/apis/submarine/v1alpha1/zz_generated.deepcopy.go
@@ -217,8 +217,25 @@ func (in *SubmarineSpec) DeepCopy() *SubmarineSpec {
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *SubmarineState) DeepCopyInto(out *SubmarineState) {
+ *out = *in
+ return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SubmarineState.
+func (in *SubmarineState) DeepCopy() *SubmarineState {
+ if in == nil {
+ return nil
+ }
+ out := new(SubmarineState)
+ in.DeepCopyInto(out)
+ return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *SubmarineStatus) DeepCopyInto(out *SubmarineStatus) {
*out = *in
+ out.SubmarineState = in.SubmarineState
return
}
diff --git a/submarine-cloud-v2/pkg/controller/controller.go b/submarine-cloud-v2/pkg/controller/controller.go
index e0bd49e..e965416 100644
--- a/submarine-cloud-v2/pkg/controller/controller.go
+++ b/submarine-cloud-v2/pkg/controller/controller.go
@@ -33,6 +33,7 @@ import (
corev1 "k8s.io/api/core/v1"
extensionsv1beta1 "k8s.io/api/extensions/v1beta1"
rbacv1 "k8s.io/api/rbac/v1"
+ "k8s.io/apimachinery/pkg/api/equality"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -84,6 +85,8 @@ const (
minioIngressRouteName = minioName + "-ingressroute"
)
+var dependents = []string{serverName, databaseName, tensorboardName, mlflowName, minioName}
+
const (
// SuccessSynced is used as part of the Event 'reason' when a Submarine is synced
SuccessSynced = "Synced"
@@ -380,102 +383,126 @@ func (c *Controller) processNextWorkItem() bool {
// syncHandler compares the actual state with the desired, and attempts to
// converge the two. It then updates the Status block of the Submarine resource
// with the current status of the resource.
+// State Machine for Submarine
+//+-----------------------------------------------------------------+
+//| +---------+ +----------+ +----------+ |
+//| | | | | | | |
+//| | New +---------> Creating +----------> Running | |
+//| | | | | | | |
+//| +----+----+ +-----+----+ +-----+----+ |
+//| | | | |
+//| | | | |
+//| | | | |
+//| | | +-----v----+ |
+//| | | | | |
+//| +--------------------+---------------> Failed | |
+//| | | |
+//| +----------+ |
+//+-----------------------------------------------------------------+
func (c *Controller) syncHandler(key string) error {
// Convert the namespace/name string into a distinct namespace and name
namespace, name, err := cache.SplitMetaNamespaceKey(key)
if err != nil {
- utilruntime.HandleError(fmt.Errorf("Invalid resource key: %s", key))
+ utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
return nil
}
klog.Info("syncHandler: ", key)
// Get the Submarine resource with this namespace/name
- submarine, err := c.submarinesLister.Submarines(namespace).Get(name)
+ submarine, err := c.getSubmarine(namespace, name)
if err != nil {
- // The Submarine resource may no longer exist, in which case we stop
- // processing
- if errors.IsNotFound(err) {
- utilruntime.HandleError(fmt.Errorf("submarine '%s' in work queue no longer exists", key))
- return nil
- }
return err
}
-
- // Submarine is in the terminating process
- if !submarine.DeletionTimestamp.IsZero() {
+ if submarine == nil {
+ // The Submarine resource may no longer exist, in which case we stop
+ // processing
+ utilruntime.HandleError(fmt.Errorf("submarine '%s' in work queue no longer exists", key))
return nil
}
- // Print out the spec of the Submarine resource
- b, err := json.MarshalIndent(submarine.Spec, "", " ")
- fmt.Println(string(b))
-
- storageType := submarine.Spec.Storage.StorageType
- if storageType != "nfs" && storageType != "host" {
- utilruntime.HandleError(fmt.Errorf("Invalid storageType '%s' found in submarine spec, nothing will be created. Valid storage types are 'nfs' and 'host'", storageType))
+ // Submarine is in the terminating process, only used when in foreground cascading deletion, otherwise the submarine will be recreated
+ if !submarine.DeletionTimestamp.IsZero() {
return nil
}
- var serverDeployment *appsv1.Deployment
- var databaseDeployment *appsv1.Deployment
+ submarineCopy := submarine.DeepCopy()
- if err != nil {
- return err
+ // Take action based on submarine state
+ switch submarineCopy.Status.SubmarineState.State {
+ case v1alpha1.NewState:
+ c.recordSubmarineEvent(submarineCopy)
+ if err := c.validateSubmarine(submarineCopy); err != nil {
+ submarineCopy.Status.SubmarineState.State = v1alpha1.FailedState
+ submarineCopy.Status.SubmarineState.ErrorMessage = err.Error()
+ c.recordSubmarineEvent(submarineCopy)
+ } else {
+ submarineCopy.Status.SubmarineState.State = v1alpha1.CreatingState
+ c.recordSubmarineEvent(submarineCopy)
+ }
+ case v1alpha1.CreatingState:
+ if err := c.createSubmarine(submarineCopy); err != nil {
+ submarineCopy.Status.SubmarineState.State = v1alpha1.FailedState
+ submarineCopy.Status.SubmarineState.ErrorMessage = err.Error()
+ c.recordSubmarineEvent(submarineCopy)
+ }
+ ok, err := c.checkSubmarineDependentsReady(submarineCopy)
+ if err != nil {
+ submarineCopy.Status.SubmarineState.State = v1alpha1.FailedState
+ submarineCopy.Status.SubmarineState.ErrorMessage = err.Error()
+ c.recordSubmarineEvent(submarineCopy)
+ }
+ if ok {
+ submarineCopy.Status.SubmarineState.State = v1alpha1.RunningState
+ c.recordSubmarineEvent(submarineCopy)
+ }
+ case v1alpha1.RunningState:
+ if err := c.createSubmarine(submarineCopy); err != nil {
+ submarineCopy.Status.SubmarineState.State = v1alpha1.FailedState
+ submarineCopy.Status.SubmarineState.ErrorMessage = err.Error()
+ c.recordSubmarineEvent(submarineCopy)
+ }
}
- serverDeployment, err = c.createSubmarineServer(submarine)
+ // update submarine status
+ err = c.updateSubmarineStatus(submarine, submarineCopy)
if err != nil {
return err
}
- databaseDeployment, err = c.createSubmarineDatabase(submarine)
- if err != nil {
- return err
- }
+ return nil
+}
- err = c.createIngress(submarine)
+func (c *Controller) updateSubmarineStatus(submarine, submarineCopy *v1alpha1.Submarine) error {
+ // Update server replicas
+ serverDeployment, err := c.getDeployment(submarine.Namespace, serverName)
if err != nil {
return err
}
-
- err = c.createSubmarineServerRBAC(submarine)
- if err != nil {
- return err
+ if serverDeployment != nil {
+ submarineCopy.Status.AvailableServerReplicas = serverDeployment.Status.AvailableReplicas
}
- err = c.createSubmarineTensorboard(submarine)
+ // Update database replicas
+ databaseDeployment, err := c.getDeployment(submarine.Namespace, databaseName)
if err != nil {
return err
}
-
- err = c.createSubmarineMlflow(submarine)
- if err != nil {
- return err
+ if databaseDeployment != nil {
+ submarineCopy.Status.AvailableDatabaseReplicas = databaseDeployment.Status.AvailableReplicas
}
- err = c.createSubmarineMinio(submarine)
- if err != nil {
- return err
+ // Skip update if nothing changed.
+ if equality.Semantic.DeepEqual(submarine.Status, submarineCopy.Status) {
+ return nil
}
- err = c.updateSubmarineStatus(submarine, serverDeployment, databaseDeployment)
+ _, err = c.submarineclientset.SubmarineV1alpha1().Submarines(submarine.Namespace).Update(context.TODO(), submarineCopy, metav1.UpdateOptions{})
if err != nil {
return err
}
-
- c.recorder.Event(submarine, corev1.EventTypeNormal, SuccessSynced, MessageResourceSynced)
-
return nil
}
-func (c *Controller) updateSubmarineStatus(submarine *v1alpha1.Submarine, serverDeployment *appsv1.Deployment, databaseDeployment *appsv1.Deployment) error {
- submarineCopy := submarine.DeepCopy()
- submarineCopy.Status.AvailableServerReplicas = serverDeployment.Status.AvailableReplicas
- submarineCopy.Status.AvailableDatabaseReplicas = databaseDeployment.Status.AvailableReplicas
- _, err := c.submarineclientset.SubmarineV1alpha1().Submarines(submarine.Namespace).Update(context.TODO(), submarineCopy, metav1.UpdateOptions{})
- return err
-}
-
// enqueueSubmarine takes a Submarine resource and converts it into a namespace/name
// string which is then put onto the work queue. This method should *not* be
// passed resources of any type other than Submarine.
@@ -531,3 +558,147 @@ func (c *Controller) handleObject(obj interface{}) {
return
}
}
+
+func (c *Controller) getSubmarine(namespace, name string) (*v1alpha1.Submarine, error) {
+ submarine, err := c.submarinesLister.Submarines(namespace).Get(name)
+ if err != nil {
+ if errors.IsNotFound(err) {
+ return nil, nil
+ }
+ return nil, err
+ }
+ return submarine, nil
+}
+
+func (c *Controller) getDeployment(namespace, name string) (*appsv1.Deployment, error) {
+ deployment, err := c.deploymentLister.Deployments(namespace).Get(name)
+ if err != nil {
+ if errors.IsNotFound(err) {
+ return nil, nil
+ }
+ return nil, err
+ }
+ return deployment, nil
+}
+
+func (c *Controller) validateSubmarine(submarine *v1alpha1.Submarine) error {
+
+ // Print out the spec of the Submarine resource
+ b, err := json.MarshalIndent(submarine.Spec, "", " ")
+ fmt.Println(string(b))
+
+ if err != nil {
+ return err
+ }
+
+ // Check storage type
+ storageType := submarine.Spec.Storage.StorageType
+ if storageType != "nfs" && storageType != "host" {
+ err = fmt.Errorf("invalid storageType '%s' found in submarine spec, nothing will be created. Valid storage types are 'nfs' and 'host'", storageType)
+ return err
+ }
+
+ return nil
+}
+
+func (c *Controller) createSubmarine(submarine *v1alpha1.Submarine) error {
+ var err error
+ err = c.createSubmarineServer(submarine)
+ if err != nil {
+ return err
+ }
+
+ err = c.createSubmarineDatabase(submarine)
+ if err != nil {
+ return err
+ }
+
+ err = c.createIngress(submarine)
+ if err != nil {
+ return err
+ }
+
+ err = c.createSubmarineServerRBAC(submarine)
+ if err != nil {
+ return err
+ }
+
+ err = c.createSubmarineTensorboard(submarine)
+ if err != nil {
+ return err
+ }
+
+ err = c.createSubmarineMlflow(submarine)
+ if err != nil {
+ return err
+ }
+
+ err = c.createSubmarineMinio(submarine)
+ if err != nil {
+ return err
+ }
+
+ return nil
+}
+
+func (c *Controller) checkSubmarineDependentsReady(submarine *v1alpha1.Submarine) (bool, error) {
+ for _, name := range dependents {
+ deployment, err := c.getDeployment(submarine.Namespace, name)
+ if err != nil {
+ return false, err
+ }
+ // check if deployment replicas failed
+ for _, condition := range deployment.Status.Conditions {
+ if condition.Type == appsv1.DeploymentReplicaFailure {
+ return false, fmt.Errorf("failed creating replicas of %s, message: %s", deployment.Name, condition.Message)
+ }
+ // progressing error, ex. ProgressDeadlineExceeded
+ if condition.Type == appsv1.DeploymentProgressing && condition.Status == corev1.ConditionFalse {
+ return false, fmt.Errorf("failed creating replicas of %s, message: %s", deployment.Name, condition.Message)
+ }
+ }
+ // check if ready replicas are same as targeted replicas
+ if deployment.Status.ReadyReplicas != deployment.Status.Replicas {
+ return false, nil
+ }
+ }
+
+ return true, nil
+}
+
+func (c *Controller) recordSubmarineEvent(submarine *v1alpha1.Submarine) {
+ switch submarine.Status.SubmarineState.State {
+ case v1alpha1.NewState:
+ c.recorder.Eventf(
+ submarine,
+ corev1.EventTypeNormal,
+ "SubmarineAdded",
+ "Submarine %s was added",
+ submarine.Name)
+ case v1alpha1.CreatingState:
+ c.recorder.Eventf(
+ submarine,
+ corev1.EventTypeNormal,
+ "SubmarineCreating",
+ "Submarine %s was creating",
+ submarine.Name,
+ )
+ case v1alpha1.RunningState:
+ c.recorder.Eventf(
+ submarine,
+ corev1.EventTypeNormal,
+ "SubmarineRunning",
+ "Submarine %s was running",
+ submarine.Name,
+ )
+ case v1alpha1.FailedState:
+ c.recorder.Eventf(
+ submarine,
+ corev1.EventTypeWarning,
+ "SubmarineFailed",
+ "Submarine %s was failed: %s",
+ submarine.Name,
+ submarine.Status.SubmarineState.ErrorMessage,
+ )
+ }
+}
diff --git a/submarine-cloud-v2/pkg/controller/submarine_database.go b/submarine-cloud-v2/pkg/controller/submarine_database.go
index 867179a..7b070ad 100644
--- a/submarine-cloud-v2/pkg/controller/submarine_database.go
+++ b/submarine-cloud-v2/pkg/controller/submarine_database.go
@@ -148,7 +148,7 @@ func newSubmarineDatabaseService(submarine *v1alpha1.Submarine) *corev1.Service
// createSubmarineDatabase is a function to create submarine-database.
// Reference: https://github.com/apache/submarine/blob/master/helm-charts/submarine/templates/submarine-database.yaml
-func (c *Controller) createSubmarineDatabase(submarine *v1alpha1.Submarine) (*appsv1.Deployment, error) {
+func (c *Controller) createSubmarineDatabase(submarine *v1alpha1.Submarine) error {
klog.Info("[createSubmarineDatabase]")
// Step 1: Create PersistentVolumeClaim
@@ -165,13 +165,13 @@ func (c *Controller) createSubmarineDatabase(submarine *v1alpha1.Submarine) (*ap
// attempt processing again later. This could have been caused by a
// temporary network failure, or any other transient reason.
if err != nil {
- return nil, err
+ return err
}
if !metav1.IsControlledBy(pvc, submarine) {
msg := fmt.Sprintf(MessageResourceExists, pvc.Name)
c.recorder.Event(submarine, corev1.EventTypeWarning, ErrResourceExists, msg)
- return nil, fmt.Errorf(msg)
+ return fmt.Errorf(msg)
}
// Step 2: Create Deployment
@@ -188,23 +188,23 @@ func (c *Controller) createSubmarineDatabase(submarine *v1alpha1.Submarine) (*ap
// attempt processing again later. This could have been caused by a
// temporary network failure, or any other transient reason.
if err != nil {
- return nil, err
+ return err
}
if !metav1.IsControlledBy(deployment, submarine) {
msg := fmt.Sprintf(MessageResourceExists, deployment.Name)
c.recorder.Event(submarine, corev1.EventTypeWarning, ErrResourceExists, msg)
- return nil, fmt.Errorf(msg)
+ return fmt.Errorf(msg)
}
// Update the replicas of the database deployment if it is not equal to spec
if submarine.Spec.Database.Replicas != nil && *submarine.Spec.Database.Replicas != *deployment.Spec.Replicas {
klog.V(4).Infof("Submarine %s database spec replicas: %d, actual replicas: %d", submarine.Name, *submarine.Spec.Database.Replicas, *deployment.Spec.Replicas)
- deployment, err = c.kubeclientset.AppsV1().Deployments(submarine.Namespace).Update(context.TODO(), newSubmarineDatabaseDeployment(submarine), metav1.UpdateOptions{})
+ _, err = c.kubeclientset.AppsV1().Deployments(submarine.Namespace).Update(context.TODO(), newSubmarineDatabaseDeployment(submarine), metav1.UpdateOptions{})
}
if err != nil {
- return nil, err
+ return err
}
// Step 3: Create Service
@@ -221,14 +221,14 @@ func (c *Controller) createSubmarineDatabase(submarine *v1alpha1.Submarine) (*ap
// attempt processing again later. This could have been caused by a
// temporary network failure, or any other transient reason.
if err != nil {
- return nil, err
+ return err
}
if !metav1.IsControlledBy(service, submarine) {
msg := fmt.Sprintf(MessageResourceExists, service.Name)
c.recorder.Event(submarine, corev1.EventTypeWarning, ErrResourceExists, msg)
- return nil, fmt.Errorf(msg)
+ return fmt.Errorf(msg)
}
- return deployment, nil
+ return nil
}
diff --git a/submarine-cloud-v2/pkg/controller/submarine_minio.go b/submarine-cloud-v2/pkg/controller/submarine_minio.go
index 0fb0121..21201b6 100644
--- a/submarine-cloud-v2/pkg/controller/submarine_minio.go
+++ b/submarine-cloud-v2/pkg/controller/submarine_minio.go
@@ -68,13 +68,13 @@ func newSubmarineMinioDeployment(submarine *v1alpha1.Submarine) *appsv1.Deployme
Spec: appsv1.DeploymentSpec{
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{
- "app": minioName + "-pod",
+ "app": minioName,
},
},
Template: corev1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: map[string]string{
- "app": minioName + "-pod",
+ "app": minioName,
},
},
Spec: corev1.PodSpec{
@@ -139,7 +139,7 @@ func newSubmarineMinioService(submarine *v1alpha1.Submarine) *corev1.Service {
Spec: corev1.ServiceSpec{
Type: corev1.ServiceTypeClusterIP,
Selector: map[string]string{
- "app": minioName + "-pod",
+ "app": minioName,
},
Ports: []corev1.ServicePort{
{
diff --git a/submarine-cloud-v2/pkg/controller/submarine_mlflow.go b/submarine-cloud-v2/pkg/controller/submarine_mlflow.go
index 33e6061..fc5f81e 100644
--- a/submarine-cloud-v2/pkg/controller/submarine_mlflow.go
+++ b/submarine-cloud-v2/pkg/controller/submarine_mlflow.go
@@ -66,13 +66,13 @@ func newSubmarineMlflowDeployment(submarine *v1alpha1.Submarine) *appsv1.Deploym
Spec: appsv1.DeploymentSpec{
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{
- "app": mlflowName + "-pod",
+ "app": mlflowName,
},
},
Template: corev1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: map[string]string{
- "app": mlflowName + "-pod",
+ "app": mlflowName,
},
},
Spec: corev1.PodSpec{
@@ -131,7 +131,7 @@ func newSubmarineMlflowService(submarine *v1alpha1.Submarine) *corev1.Service {
Spec: corev1.ServiceSpec{
Type: corev1.ServiceTypeClusterIP,
Selector: map[string]string{
- "app": mlflowName + "-pod",
+ "app": mlflowName,
},
Ports: []corev1.ServicePort{
{
diff --git a/submarine-cloud-v2/pkg/controller/submarine_server.go b/submarine-cloud-v2/pkg/controller/submarine_server.go
index 10ac5e3..b2c9988 100644
--- a/submarine-cloud-v2/pkg/controller/submarine_server.go
+++ b/submarine-cloud-v2/pkg/controller/submarine_server.go
@@ -47,7 +47,7 @@ func newSubmarineServerService(submarine *v1alpha1.Submarine) *corev1.Service {
ObjectMeta: metav1.ObjectMeta{
Name: serverName,
Labels: map[string]string{
- "run": serverName,
+ "app": serverName,
},
OwnerReferences: []metav1.OwnerReference{
*metav1.NewControllerRef(submarine, v1alpha1.SchemeGroupVersion.WithKind("Submarine")),
@@ -62,7 +62,7 @@ func newSubmarineServerService(submarine *v1alpha1.Submarine) *corev1.Service {
},
},
Selector: map[string]string{
- "run": serverName,
+ "app": serverName,
},
},
}
@@ -87,14 +87,14 @@ func newSubmarineServerDeployment(submarine *v1alpha1.Submarine) *appsv1.Deploym
Spec: appsv1.DeploymentSpec{
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{
- "run": serverName,
+ "app": serverName,
},
},
Replicas: &serverReplicas,
Template: corev1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: map[string]string{
- "run": serverName,
+ "app": serverName,
},
},
Spec: corev1.PodSpec{
@@ -157,7 +157,7 @@ func newSubmarineServerDeployment(submarine *v1alpha1.Submarine) *appsv1.Deploym
// createSubmarineServer is a function to create submarine-server.
// Reference: https://github.com/apache/submarine/blob/master/helm-charts/submarine/templates/submarine-server.yaml
-func (c *Controller) createSubmarineServer(submarine *v1alpha1.Submarine) (*appsv1.Deployment, error) {
+func (c *Controller) createSubmarineServer(submarine *v1alpha1.Submarine) error {
klog.Info("[createSubmarineServer]")
// Step1: Create ServiceAccount
@@ -172,13 +172,13 @@ func (c *Controller) createSubmarineServer(submarine *v1alpha1.Submarine) (*apps
// attempt processing again later. This could have been caused by a
// temporary network failure, or any other transient reason.
if err != nil {
- return nil, err
+ return err
}
if !metav1.IsControlledBy(serviceaccount, submarine) {
msg := fmt.Sprintf(MessageResourceExists, serviceaccount.Name)
c.recorder.Event(submarine, corev1.EventTypeWarning, ErrResourceExists, msg)
- return nil, fmt.Errorf(msg)
+ return fmt.Errorf(msg)
}
// Step2: Create Service
@@ -193,13 +193,13 @@ func (c *Controller) createSubmarineServer(submarine *v1alpha1.Submarine) (*apps
// attempt processing again later. This could have been caused by a
// temporary network failure, or any other transient reason.
if err != nil {
- return nil, err
+ return err
}
if !metav1.IsControlledBy(service, submarine) {
msg := fmt.Sprintf(MessageResourceExists, service.Name)
c.recorder.Event(submarine, corev1.EventTypeWarning, ErrResourceExists, msg)
- return nil, fmt.Errorf(msg)
+ return fmt.Errorf(msg)
}
// Step3: Create Deployment
@@ -214,24 +214,24 @@ func (c *Controller) createSubmarineServer(submarine *v1alpha1.Submarine) (*apps
// attempt processing again later. This could have been caused by a
// temporary network failure, or any other transient reason.
if err != nil {
- return nil, err
+ return err
}
if !metav1.IsControlledBy(deployment, submarine) {
msg := fmt.Sprintf(MessageResourceExists, deployment.Name)
c.recorder.Event(submarine, corev1.EventTypeWarning, ErrResourceExists, msg)
- return nil, fmt.Errorf(msg)
+ return fmt.Errorf(msg)
}
// Update the replicas of the server deployment if it is not equal to spec
if submarine.Spec.Server.Replicas != nil && *submarine.Spec.Server.Replicas != *deployment.Spec.Replicas {
klog.V(4).Infof("Submarine %s server spec replicas: %d, actual replicas: %d", submarine.Name, *submarine.Spec.Server.Replicas, *deployment.Spec.Replicas)
- deployment, err = c.kubeclientset.AppsV1().Deployments(submarine.Namespace).Update(context.TODO(), newSubmarineServerDeployment(submarine), metav1.UpdateOptions{})
+ _, err = c.kubeclientset.AppsV1().Deployments(submarine.Namespace).Update(context.TODO(), newSubmarineServerDeployment(submarine), metav1.UpdateOptions{})
}
if err != nil {
- return nil, err
+ return err
}
- return deployment, nil
+ return nil
}
diff --git a/submarine-cloud-v2/pkg/controller/submarine_tensorboard.go b/submarine-cloud-v2/pkg/controller/submarine_tensorboard.go
index e2a584e..5dd3aad 100644
--- a/submarine-cloud-v2/pkg/controller/submarine_tensorboard.go
+++ b/submarine-cloud-v2/pkg/controller/submarine_tensorboard.go
@@ -68,13 +68,13 @@ func newSubmarineTensorboardDeployment(submarine *v1alpha1.Submarine) *appsv1.De
Spec: appsv1.DeploymentSpec{
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{
- "app": tensorboardName + "-pod",
+ "app": tensorboardName,
},
},
Template: corev1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: map[string]string{
- "app": tensorboardName + "-pod",
+ "app": tensorboardName,
},
},
Spec: corev1.PodSpec{
@@ -136,7 +136,7 @@ func newSubmarineTensorboardService(submarine *v1alpha1.Submarine) *corev1.Servi
},
Spec: corev1.ServiceSpec{
Selector: map[string]string{
- "app": tensorboardName + "-pod",
+ "app": tensorboardName,
},
Ports: []corev1.ServicePort{
{
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@submarine.apache.org
For additional commands, e-mail: dev-help@submarine.apache.org