You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@submarine.apache.org by GitBox <gi...@apache.org> on 2021/08/20 16:05:52 UTC
[GitHub] [submarine] Kenchu123 opened a new pull request #719: SUBMARINE-950. Create state machine for submarine custom resource
Kenchu123 opened a new pull request #719:
URL: https://github.com/apache/submarine/pull/719
### 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@submarine.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [submarine] kevin85421 commented on pull request #719: SUBMARINE-950. Create state machine for submarine custom resource
Posted by GitBox <gi...@apache.org>.
kevin85421 commented on pull request #719:
URL: https://github.com/apache/submarine/pull/719#issuecomment-906475271
LGTM
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@submarine.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [submarine] Kenchu123 commented on a change in pull request #719: SUBMARINE-950. Create state machine for submarine custom resource
Posted by GitBox <gi...@apache.org>.
Kenchu123 commented on a change in pull request #719:
URL: https://github.com/apache/submarine/pull/719#discussion_r696298282
##########
File path: submarine-cloud-v2/pkg/controller/controller.go
##########
@@ -531,3 +561,149 @@ 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 {
+ // The Submarine resource may no longer exist, in which case we stop
+ // processing
Review comment:
Ok, thanks.
##########
File path: submarine-cloud-v2/pkg/controller/controller.go
##########
@@ -380,102 +383,129 @@ 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 {
+ return err
+ }
+ if submarine == 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
+ utilruntime.HandleError(fmt.Errorf("submarine '%s' in work queue no longer exists", key))
+ return nil
}
// Submarine is in the terminating process
if !submarine.DeletionTimestamp.IsZero() {
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))
- 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)
- if err != nil {
- return err
+ // update submarine status
+ if submarineCopy != nil {
Review comment:
Ok, thanks.
##########
File path: submarine-cloud-v2/pkg/controller/controller.go
##########
@@ -531,3 +561,149 @@ 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 {
+ // The Submarine resource may no longer exist, in which case we stop
+ // processing
+ 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" {
+ utilruntime.HandleError(fmt.Errorf("invalid storageType '%s' found in submarine spec, nothing will be created. Valid storage types are 'nfs' and 'host'", storageType))
+ return nil
+ }
Review comment:
Ok, thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@submarine.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [submarine] Kenchu123 commented on a change in pull request #719: SUBMARINE-950. Create state machine for submarine custom resource
Posted by GitBox <gi...@apache.org>.
Kenchu123 commented on a change in pull request #719:
URL: https://github.com/apache/submarine/pull/719#discussion_r696273895
##########
File path: submarine-cloud-v2/pkg/controller/controller.go
##########
@@ -380,102 +383,129 @@ 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))
Review comment:
There is a warning: `Error strings should not be capitalized (unless beginning with proper nouns or acronyms) or end with punctuation`.
Ref. https://github.com/golang/go/wiki/CodeReviewComments#error-strings
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@submarine.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [submarine] Kenchu123 commented on a change in pull request #719: SUBMARINE-950. Create state machine for submarine custom resource
Posted by GitBox <gi...@apache.org>.
Kenchu123 commented on a change in pull request #719:
URL: https://github.com/apache/submarine/pull/719#discussion_r696274569
##########
File path: submarine-cloud-v2/pkg/controller/controller.go
##########
@@ -380,102 +383,129 @@ 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 {
+ return err
+ }
+ if submarine == 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
+ utilruntime.HandleError(fmt.Errorf("submarine '%s' in work queue no longer exists", key))
+ return nil
}
// Submarine is in the terminating process
if !submarine.DeletionTimestamp.IsZero() {
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))
- 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)
- if err != nil {
- return err
+ // update submarine status
+ if submarineCopy != nil {
+ err = c.updateSubmarineStatus(submarine, submarineCopy)
+ if err != nil {
+ return err
+ }
}
- databaseDeployment, err = c.createSubmarineDatabase(submarine)
- if err != nil {
- return err
- }
+ // c.recorder.Event(submarine, corev1.EventTypeNormal, SuccessSynced, MessageResourceSynced)
Review comment:
I don't find a need to record this event.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@submarine.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [submarine] Kenchu123 commented on a change in pull request #719: SUBMARINE-950. Create state machine for submarine custom resource
Posted by GitBox <gi...@apache.org>.
Kenchu123 commented on a change in pull request #719:
URL: https://github.com/apache/submarine/pull/719#discussion_r696298973
##########
File path: submarine-cloud-v2/pkg/controller/controller.go
##########
@@ -531,3 +561,149 @@ 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 {
+ // The Submarine resource may no longer exist, in which case we stop
+ // processing
+ 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" {
+ utilruntime.HandleError(fmt.Errorf("invalid storageType '%s' found in submarine spec, nothing will be created. Valid storage types are 'nfs' and 'host'", storageType))
+ return nil
+ }
+
+ 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 {
+ podList, err := c.kubeclientset.CoreV1().Pods(submarine.Namespace).List(context.TODO(), metav1.ListOptions{LabelSelector: fmt.Sprintf("app=%s", name)})
+ if err != nil {
+ return false, err
+ }
+ for _, pod := range podList.Items {
+ switch pod.Status.Phase {
+ case corev1.PodPending:
+ return false, nil
+ case corev1.PodFailed, corev1.PodSucceeded:
+ return false, fmt.Errorf("pod completed")
+ case corev1.PodRunning:
+ for _, condition := range pod.Status.Conditions {
+ if condition.Type == corev1.PodReady && condition.Status != corev1.ConditionTrue {
+ return false, nil
+ }
+ }
+ }
+ }
+ }
+
+ return true, nil
+}
Review comment:
Thanks. I didn't know about the deployment status before. I will check it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@submarine.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [submarine] MortalHappiness commented on a change in pull request #719: SUBMARINE-950. Create state machine for submarine custom resource
Posted by GitBox <gi...@apache.org>.
MortalHappiness commented on a change in pull request #719:
URL: https://github.com/apache/submarine/pull/719#discussion_r695507150
##########
File path: submarine-cloud-v2/pkg/controller/controller.go
##########
@@ -380,102 +383,129 @@ 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))
Review comment:
Why this "Invalid" needs to be uncapitalized?
##########
File path: submarine-cloud-v2/pkg/controller/controller.go
##########
@@ -531,3 +561,149 @@ 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 {
+ // The Submarine resource may no longer exist, in which case we stop
+ // processing
+ 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" {
+ utilruntime.HandleError(fmt.Errorf("invalid storageType '%s' found in submarine spec, nothing will be created. Valid storage types are 'nfs' and 'host'", storageType))
+ return nil
+ }
+
+ 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 {
+ podList, err := c.kubeclientset.CoreV1().Pods(submarine.Namespace).List(context.TODO(), metav1.ListOptions{LabelSelector: fmt.Sprintf("app=%s", name)})
+ if err != nil {
+ return false, err
+ }
+ for _, pod := range podList.Items {
+ switch pod.Status.Phase {
+ case corev1.PodPending:
+ return false, nil
+ case corev1.PodFailed, corev1.PodSucceeded:
+ return false, fmt.Errorf("pod completed")
+ case corev1.PodRunning:
+ for _, condition := range pod.Status.Conditions {
+ if condition.Type == corev1.PodReady && condition.Status != corev1.ConditionTrue {
+ return false, nil
+ }
+ }
+ }
+ }
+ }
+
+ return true, nil
+}
Review comment:
Why do you need to check the running state for all the pods belonging to the deployments instead of simply checking the status of the deployments? Is there any benefit? For example the [ReadyReplicas in DeploymentStatus](https://pkg.go.dev/k8s.io/api/apps/v1#DeploymentStatus).
##########
File path: submarine-cloud-v2/pkg/controller/controller.go
##########
@@ -531,3 +561,149 @@ 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 {
+ // The Submarine resource may no longer exist, in which case we stop
+ // processing
Review comment:
These comments can be removed.
##########
File path: submarine-cloud-v2/pkg/controller/controller.go
##########
@@ -380,102 +383,129 @@ 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 {
+ return err
+ }
+ if submarine == 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
+ utilruntime.HandleError(fmt.Errorf("submarine '%s' in work queue no longer exists", key))
+ return nil
}
// Submarine is in the terminating process
if !submarine.DeletionTimestamp.IsZero() {
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))
- 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)
- if err != nil {
- return err
+ // update submarine status
+ if submarineCopy != nil {
Review comment:
`submarineCopy` will never be `nil`, so this `if` statement is redundant.
##########
File path: submarine-cloud-v2/pkg/controller/controller.go
##########
@@ -380,102 +383,129 @@ 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 {
+ return err
+ }
+ if submarine == 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
Review comment:
These lines should be preserved. See the comments on lines 568 to 572.
##########
File path: submarine-cloud-v2/pkg/controller/controller.go
##########
@@ -380,102 +383,129 @@ 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 {
+ return err
+ }
+ if submarine == 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
+ utilruntime.HandleError(fmt.Errorf("submarine '%s' in work queue no longer exists", key))
+ return nil
}
// Submarine is in the terminating process
if !submarine.DeletionTimestamp.IsZero() {
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))
- 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)
- if err != nil {
- return err
+ // update submarine status
+ if submarineCopy != nil {
+ err = c.updateSubmarineStatus(submarine, submarineCopy)
+ if err != nil {
+ return err
+ }
}
- databaseDeployment, err = c.createSubmarineDatabase(submarine)
- if err != nil {
- return err
- }
+ // c.recorder.Event(submarine, corev1.EventTypeNormal, SuccessSynced, MessageResourceSynced)
Review comment:
Why this line should be commented out?
##########
File path: submarine-cloud-v2/pkg/controller/controller.go
##########
@@ -531,3 +561,149 @@ 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 {
+ // The Submarine resource may no longer exist, in which case we stop
+ // processing
+ 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" {
+ utilruntime.HandleError(fmt.Errorf("invalid storageType '%s' found in submarine spec, nothing will be created. Valid storage types are 'nfs' and 'host'", storageType))
+ return nil
+ }
Review comment:
This is a fatal error and the submarine should enter the failed state. Change these lines to the following code.
```go
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
}
```
##########
File path: submarine-cloud-v2/pkg/controller/controller.go
##########
@@ -380,102 +383,129 @@ 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 {
+ return err
+ }
+ if submarine == 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
+ utilruntime.HandleError(fmt.Errorf("submarine '%s' in work queue no longer exists", key))
+ return nil
}
// Submarine is in the terminating process
Review comment:
Can you add more comments for this line? For example, "Submarine is in the terminating process, only used when cascading-delete=foreground, otherwise the submarine will be recreated"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@submarine.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [submarine] MortalHappiness commented on a change in pull request #719: SUBMARINE-950. Create state machine for submarine custom resource
Posted by GitBox <gi...@apache.org>.
MortalHappiness commented on a change in pull request #719:
URL: https://github.com/apache/submarine/pull/719#discussion_r696351140
##########
File path: submarine-cloud-v2/pkg/controller/controller.go
##########
@@ -380,102 +383,129 @@ 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))
Review comment:
Ok. Thanks. I got the point.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@submarine.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [submarine] Kenchu123 commented on pull request #719: SUBMARINE-950. Create state machine for submarine custom resource
Posted by GitBox <gi...@apache.org>.
Kenchu123 commented on pull request #719:
URL: https://github.com/apache/submarine/pull/719#issuecomment-906109541
@MortalHappiness. I have modified the code. Can you help me review it again?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@submarine.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [submarine] Kenchu123 commented on a change in pull request #719: SUBMARINE-950. Create state machine for submarine custom resource
Posted by GitBox <gi...@apache.org>.
Kenchu123 commented on a change in pull request #719:
URL: https://github.com/apache/submarine/pull/719#discussion_r696298522
##########
File path: submarine-cloud-v2/pkg/controller/controller.go
##########
@@ -380,102 +383,129 @@ 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 {
+ return err
+ }
+ if submarine == 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
+ utilruntime.HandleError(fmt.Errorf("submarine '%s' in work queue no longer exists", key))
+ return nil
}
// Submarine is in the terminating process
Review comment:
Ok, thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@submarine.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [submarine] asfgit closed pull request #719: SUBMARINE-950. Create state machine for submarine custom resource
Posted by GitBox <gi...@apache.org>.
asfgit closed pull request #719:
URL: https://github.com/apache/submarine/pull/719
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@submarine.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org