feat: support for grow and shrink

This is the first fully working design (and example) for
growing the cluster based on ensemble rules! Very awesome!

Signed-off-by: vsoch <[email protected]>

api/v1alpha1/ensemble_types.go

@@ -102,9 +102,9 @@ func (m *Member) Size() int32 {
 	return 0
 }
 
-//func (e *Ensemble) RequeueAfter() time.Duration {
-//	return time.Duration(time.Duration(e.Spec.CheckSeconds) * time.Second)
-//}
+func (e *Ensemble) ServiceName() string {
+	return fmt.Sprintf("%s-grpc", e.Name)
+}
 
 // Validate ensures we have data that is needed, and sets defaults if needed
 func (e *Ensemble) Validate() error {

controllers/ensemble/api.go

@@ -20,11 +20,9 @@ import (
 )
 
 // getDeploymentAddress gets the address of the deployment
-// Note that this assumes we have one container running (and one address)
-func (r *EnsembleReconciler) getServiceAddress(
+func (r *EnsembleReconciler) getDeploymentAddress(
 	ctx context.Context,
 	ensemble *api.Ensemble,
-	name string,
 ) (string, error) {
 
 	// The MiniCluster service is being provided by the index 0 pod, so we can find it here.
@@ -73,6 +71,44 @@ func (r *EnsembleReconciler) getServiceAddress(
 	return ipAddress, nil
 }
 
+// getServiceAddress gets the service ClusterIP serving the grpc endpoint
+func (r *EnsembleReconciler) getServiceAddress(
+	ctx context.Context,
+	ensemble *api.Ensemble,
+) (string, error) {
+
+	// The MiniCluster service is being provided by the index 0 pod, so we can find it here.
+	clientset, err := kubernetes.NewForConfig(r.RESTConfig)
+	if err != nil {
+		return "", err
+	}
+
+	// List all services with this name (just the one!)
+	services, err := clientset.CoreV1().Services(ensemble.Namespace).List(
+		ctx,
+		metav1.ListOptions{
+			FieldSelector: "metadata.name=" + ensemble.ServiceName(),
+		},
+	)
+	if err != nil {
+		return "", err
+	}
+
+	// Get the ip address of the first (only for now) pod
+	var ipAddress string
+	for _, svc := range services.Items {
+		ipAddress = svc.Spec.ClusterIP
+		break
+	}
+
+	// If we don't have an ip address yet, try again later
+	if ipAddress == "" {
+		fmt.Println("      No grpc services found")
+		return "", fmt.Errorf("no grpc services found, not ready yet")
+	}
+	return ipAddress, nil
+}
+
 func (r *EnsembleReconciler) createServiceAccount(
 	ctx context.Context,
 	ensemble *api.Ensemble,
@@ -146,10 +182,11 @@ func (r *EnsembleReconciler) createRole(
 					{
 						APIGroups: []string{"flux-framework.org"},
 						Resources: []string{"miniclusters"},
-						Verbs:     []string{"get", "list", "create", "update", "delete"},
+						Verbs:     []string{"get", "list", "create", "update", "delete", "patch"},
 					},
 				},
 			}
+
 			ctrl.SetControllerReference(ensemble, role, r.Scheme)
 			err = r.Create(ctx, role)
 			if err != nil {
@@ -212,37 +249,42 @@ func (r *EnsembleReconciler) createRoleBinding(
 
 }
 
+// createService creates the service for the grpc
+// This is used to expose the port to the cluster
+// TODO stopped here - bring up interactive and debug grpc (it worked before)
 func (r *EnsembleReconciler) createService(
 	ctx context.Context,
 	ensemble *api.Ensemble,
 ) (ctrl.Result, error) {
 
-	serviceName := fmt.Sprintf("%s-grpc", ensemble.Name)
-
 	// First see if we already have it!
 	svc := &corev1.Service{}
 	err := r.Get(
 		ctx,
 		types.NamespacedName{
-			Name:      serviceName,
+			Name:      ensemble.ServiceName(),
 			Namespace: ensemble.Namespace,
 		},
 		svc,
 	)
 
-	// Deployment labels to match for service
-	appLabels := getDeploymentLabels(ensemble)
-	port, err := strconv.Atoi(ensemble.Spec.Sidecar.Port)
-	if err != nil {
-		return ctrl.Result{}, err
-	}
-
 	// If we haven't found it, create it
 	if err != nil {
 		if errors.IsNotFound(err) {
+
+			// Deployment labels to match for service
+			appLabels := getDeploymentLabels(ensemble)
+			port, err := strconv.Atoi(ensemble.Spec.Sidecar.Port)
+			if err != nil {
+				return ctrl.Result{}, err
+			}
+
 			svc = &corev1.Service{
-				TypeMeta:   metav1.TypeMeta{},
-				ObjectMeta: metav1.ObjectMeta{Name: serviceName, Namespace: ensemble.Namespace},
+				TypeMeta: metav1.TypeMeta{},
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      ensemble.ServiceName(),
+					Namespace: ensemble.Namespace,
+				},
 				Spec: corev1.ServiceSpec{
 					Ports: []corev1.ServicePort{
 						{
@@ -271,7 +313,7 @@ func (r *EnsembleReconciler) createService(
 	}
 	// We already have the service account, no error
 	// and continue to next thing.
-	return ctrl.Result{}, nil
+	return ctrl.Result{Requeue: true}, nil
 }
 
 // ensureEnsembleService creates the deployment to run the ensemble service
@@ -389,6 +431,7 @@ func (r *EnsembleReconciler) newEnsembleDeployment(ensemble *api.Ensemble) (*app
 	command := []string{
 		"ensemble-server",
 		"start",
+		"--kubernetes",
 		"--host", "0.0.0.0",
 		"--port", ensemble.Spec.Sidecar.Port,
 		"--workers", workers,
@@ -411,7 +454,9 @@ func (r *EnsembleReconciler) newEnsembleDeployment(ensemble *api.Ensemble) (*app
 					Labels: appLabels,
 				},
 				Spec: corev1.PodSpec{
-					Subdomain:          ensemble.Name,
+
+					// This needs to match the service name
+					Subdomain:          ensemble.ServiceName(),
 					ServiceAccountName: ensemble.Name,
 					Containers: []corev1.Container{
 						{

controllers/ensemble/configmap.go

@@ -39,6 +39,7 @@ var (
 // getConfigMap gets the entrypoint config map
 func (r *EnsembleReconciler) ensureEnsembleConfig(
 	ctx context.Context,
+	name string,
 	ensemble *api.Ensemble,
 	member *api.Member,
 ) (ctrl.Result, error) {
@@ -49,7 +50,7 @@ func (r *EnsembleReconciler) ensureEnsembleConfig(
 	err := r.Get(
 		ctx,
 		types.NamespacedName{
-			Name:      ensemble.Name,
+			Name:      name,
 			Namespace: ensemble.Namespace,
 		},
 		existing,
@@ -61,7 +62,7 @@ func (r *EnsembleReconciler) ensureEnsembleConfig(
 		if errors.IsNotFound(err) {
 
 			// Finally create the config map
-			cm := r.createConfigMap(ensemble, member)
+			cm := r.createConfigMap(ensemble, member, name)
 			r.Log.Info("✨ Creating Ensemble YAML ✨")
 			err = r.Create(ctx, cm)
 			if err != nil {
@@ -80,15 +81,19 @@ func (r *EnsembleReconciler) ensureEnsembleConfig(
 }
 
 // createConfigMap generates a config map with some kind of data
-func (r *EnsembleReconciler) createConfigMap(ensemble *api.Ensemble, member *api.Member) *corev1.ConfigMap {
+func (r *EnsembleReconciler) createConfigMap(
+	ensemble *api.Ensemble,
+	member *api.Member,
+	name string,
+) *corev1.ConfigMap {
 
 	data := map[string]string{
 		ensembleYamlName: member.Ensemble,
 	}
 	cm := &corev1.ConfigMap{
 		TypeMeta: metav1.TypeMeta{},
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      ensemble.Name,
+			Name:      name,
 			Namespace: ensemble.Namespace,
 		},
 		Data: data,

controllers/ensemble/ensemble_controller.go

@@ -122,15 +122,16 @@ func (r *EnsembleReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		// This indicates the ensemble member is a MiniCluster
 		if !reflect.DeepEqual(member.MiniCluster, minicluster.MiniClusterSpec{}) {
 
+			// Name is the index + ensemble name
+			name := fmt.Sprintf("%s-%d", ensemble.Name, i)
+
 			// Create the config map volume (the ensemble.yaml)
 			// for the MiniCluster to run as the entrypoint
-			result, err := r.ensureEnsembleConfig(ctx, &ensemble, &member)
+			result, err := r.ensureEnsembleConfig(ctx, name, &ensemble, &member)
 			if err != nil {
 				return result, err
 			}
 
-			// Name is the index + ensemble name
-			name := fmt.Sprintf("%s-%d", ensemble.Name, i)
 			result, err = r.ensureMiniClusterEnsemble(ctx, name, &ensemble, &member)
 			if err != nil {
 				return result, err

controllers/ensemble/minicluster.go

@@ -52,7 +52,7 @@ func (r *EnsembleReconciler) ensureMiniClusterEnsemble(
 
 			// We first need the address of the grpc service
 			// if this fails, we try again - it might not be ready
-			ipAddress, err := r.getServiceAddress(ctx, ensemble, name)
+			ipAddress, err := r.getServiceAddress(ctx, ensemble)
 			if err != nil {
 				return ctrl.Result{Requeue: true}, err
 			}
@@ -96,45 +96,6 @@ func (r *EnsembleReconciler) getExistingMiniCluster(
 	return existing, err
 }
 
-// updateMiniCluster size gets its current size from the status and updated
-// if it is valid
-func (r *EnsembleReconciler) updateMiniClusterSize(
-	ctx context.Context,
-	ensemble *api.Ensemble,
-	scale int32,
-	name string,
-) (ctrl.Result, error) {
-
-	mc, err := r.getExistingMiniCluster(ctx, name, ensemble)
-
-	// Check the size against what we have
-	size := mc.Spec.Size
-
-	// We can only scale if we are left with at least one node
-	// If we want to scale to 0, this should be a termination event
-	newSize := size + scale
-	if newSize < 1 {
-		fmt.Printf("        Ignoring scaling event, new size %d is < 1\n", newSize)
-		return ctrl.Result{}, err
-	}
-	if newSize <= mc.Spec.MaxSize {
-		fmt.Printf("        Updating size from %d to %d\n", size, newSize)
-		mc.Spec.Size = newSize
-
-		// TODO: this will trigger reconcile. Can we set the time?
-		err = r.Update(ctx, mc)
-		if err != nil {
-			return ctrl.Result{}, err
-		}
-
-	} else {
-		fmt.Printf("        Ignoring scaling event %d to %d, outside allowed boundary\n", size, newSize)
-	}
-
-	// Check again in the allotted time
-	return ctrl.Result{}, err
-}
-
 // newMiniCluster creates a new ensemble minicluster
 func (r *EnsembleReconciler) newMiniCluster(
 	name string,
@@ -159,11 +120,11 @@ func (r *EnsembleReconciler) newMiniCluster(
 	// Add the config map as a volume to the main container
 	container := spec.Spec.Containers[0]
 	volume := minicluster.ContainerVolume{
-		ConfigMapName: ensemble.Name,
+		ConfigMapName: name,
 		Path:          "/ensemble-entrypoint",
 		Items:         items,
 	}
-	container.Volumes = map[string]minicluster.ContainerVolume{ensemble.Name: volume}
+	container.Volumes = map[string]minicluster.ContainerVolume{name: volume}
 	container.RunFlux = true
 	container.Launcher = true
 
@@ -177,10 +138,10 @@ func (r *EnsembleReconciler) newMiniCluster(
 	// Note that we aren't creating a headless service so that the different members are isolated.
 	// Otherwise they would all be on the same service address, which might get ugly.
 	ensembleYamlPath := filepath.Join(ensembleYamlDirName, ensembleYamlName)
-	prefix := "ensemble run --executor minicluster --host"
+	prefix := "ensemble run --kubernetes --executor minicluster --host"
 	container.Command = fmt.Sprintf("%s %s --port %s --name %s %s",
 		prefix, host,
-		ensemble.Spec.Sidecar.Port, ensemble.Name,
+		ensemble.Spec.Sidecar.Port, name,
 		ensembleYamlPath,
 	)
 	spec.Spec.Containers[0] = container

docs/getting_started/custom-resource-definition.md

@@ -129,8 +129,7 @@ spec:
 #### Sidecar
 
 The sidecar is where the gRPC service (deployment) runs alongside the members. You can customize options related
-to this deployment, although you likely don't need to. I find this useful for development (e.g., using a development container
-and asking to pull always). These are the options available:
+to this deployment, although you likely don't need to. I find this useful for development (e.g., using a development container and asking to pull always). These are the options available:
 
 
 ```yaml
@@ -163,7 +162,6 @@ Members is a list of members to add to your ensemble. In the future this could s
 but for now we are focusing on Flux Operator MiniCluster, which has a nice setup to allow for a sidecar container
 to monitor the Flux queue, doing everything from submitting jobs to reporting status. This is a list, so you
 could have two MiniCluster types, for example, that have different resources. For each member, you can define the following:
-
 ##### Ensemble
 
 The ensemble section is a text chunk that should coincide with the ensemble.yaml that is described by ensemble-python. It will create a config map that is mapped as a volume to run the ensemble.
@@ -178,3 +176,15 @@ start the MiniCluster in interactive mode.
 
 Note that for sidecar images, we provide automated builds for two versions of each of rocky and ubuntu.
 You can find them [here](https://github.com/converged-computing/ensemble-operator/pkgs/container/ensemble-operator-api).
+
+##### Branch
+
+If you want to test a development branch of ensemble-python, you can specify it alongside your minicluster / ensemble.
+For example:
+
+```yaml
+  # Install ensemble python from this branch instead of pip (for development)
+  - branch: add-support-minicluster-autoscale
+    minicluster:
+      ...
+```