Feature/gh 77 scale a container on gke

CHANGELOG.md

@@ -9,6 +9,7 @@
 ### ENHANCEMENTS
 
 * Concurrent workflows and custom commands executions are now allowed except when a deployment/undeployment/scaling operation is in progress ([GH-182](https://github.com/ystia/yorc/issues/182))
+* Enable scaling of Kubernetes deployments ([GH-77](https://github.com/ystia/yorc/issues/77))
 
 ## 3.1.0-M4 (October 08, 2018)
 

data/tosca/yorc-kubernetes-types.yml

@@ -20,7 +20,22 @@ artifact_types:
 node_types:
   yorc.nodes.kubernetes.api.types.DeploymentResource:
     derived_from: org.alien4cloud.kubernetes.api.types.DeploymentResource
+    attributes:
+      replicas:
+        type: integer
+        description: >
+          Current number of replicas for this deployment
     interfaces:
+      org.alien4cloud.management.ClusterControl:
+        scale:
+          inputs:
+            EXPECTED_INSTANCES:
+              type: integer
+            INSTANCES_DELTA:
+              type: integer
+          implementation:
+            file: "embedded"
+            type: yorc.artifacts.Deployment.Kubernetes
       Standard:
         create:
           implementation:

prov/kubernetes/execution.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"net/url"
 	"path"
+	"strconv"
 	"strings"
 	"time"
 
@@ -54,6 +55,7 @@ type k8sResourceOperation int
 const (
 	k8sCreateOperation k8sResourceOperation = iota
 	k8sDeleteOperation
+	k8sScaleOperation
 )
 
 type execution interface {
@@ -151,6 +153,8 @@ func (e *executionCommon) execute(ctx context.Context, clientset *kubernetes.Cli
 		return e.uninstallNode(ctx, clientset)
 	case "standard.delete":
 		return e.manageKubernetesResource(ctx, clientset, generator, k8sDeleteOperation)
+	case "org.alien4cloud.management.clustercontrol.scale":
+		return e.manageKubernetesResource(ctx, clientset, generator, k8sScaleOperation)
 	default:
 		return errors.Errorf("Unsupported operation %q", e.Operation.Name)
 	}
@@ -248,7 +252,10 @@ func (e *executionCommon) manageDeploymentResource(ctx context.Context, clientse
 		if err != nil {
 			return err
 		}
-
+		err = deployments.SetAttributeForAllInstances(e.kv, e.deploymentID, e.NodeName, "replicas", fmt.Sprint(*deployment.Spec.Replicas))
+		if err != nil {
+			return err
+		}
 		events.WithContextOptionalFields(ctx).NewLogEntry(events.LogLevelDEBUG, e.deploymentID).Registerf("k8s Deployment %s created in namespace %s", deployment.Name, namespace)
 	case k8sDeleteOperation:
 		// Delete Deployment k8s resource
@@ -286,7 +293,33 @@ func (e *executionCommon) manageDeploymentResource(ctx context.Context, clientse
 			return err
 		}
 		events.WithContextOptionalFields(ctx).NewLogEntry(events.LogLevelINFO, e.deploymentID).Registerf("k8s Namespace %s deleted", namespace)
+	case k8sScaleOperation:
+		expectedInstances, err := tasks.GetTaskInput(e.kv, e.taskID, "EXPECTED_INSTANCES")
+		if err != nil {
+			return err
+		}
+		r, err := strconv.ParseInt(expectedInstances, 10, 32)
+		if err != nil {
+			return errors.Wrapf(err, "failed to parse EXPECTED_INSTANCES: %q parameter as integer", expectedInstances)
+		}
+		replicas := int32(r)
+		deploymentRepr.Spec.Replicas = &replicas
+
+		deployment, err := clientset.ExtensionsV1beta1().Deployments(namespace).Update(&deploymentRepr)
+		if err != nil {
+			return errors.Wrap(err, "failed to update kubernetes deployment for scaling")
+		}
+		streamDeploymentLogs(ctx, e.deploymentID, clientset, deployment)
 
+		err = waitForDeploymentCompletion(ctx, e.deploymentID, clientset, deployment)
+		if err != nil {
+			return err
+		}
+		err = deployments.SetAttributeForAllInstances(e.kv, e.deploymentID, e.NodeName, "replicas", expectedInstances)
+		if err != nil {
+			return err
+		}
+		events.WithContextOptionalFields(ctx).NewLogEntry(events.LogLevelDEBUG, e.deploymentID).Registerf("k8s Deployment %s scaled to %s instances in namespace %s", deployment.Name, expectedInstances, namespace)
 	default:
 		return errors.Errorf("Unsupported operation on k8s resource")
 	}

rest/dep_custom.go

@@ -15,18 +15,19 @@
 package rest
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"net/http"
 	"path"
-	"strconv"
+
+	"github.com/ystia/yorc/prov/operations"
 
 	"strings"
 
 	"github.com/julienschmidt/httprouter"
 	"github.com/ystia/yorc/deployments"
-	"github.com/ystia/yorc/helper/consulutil"
 	"github.com/ystia/yorc/log"
 	"github.com/ystia/yorc/tasks"
 )
@@ -51,28 +52,30 @@ func (s *Server) newCustomCommandHandler(w http.ResponseWriter, r *http.Request)
 		log.Panic(err)
 	}
 
-	var inputMap CustomCommandRequest
-	if err = json.Unmarshal(body, &inputMap); err != nil {
+	var ccRequest CustomCommandRequest
+	if err = json.Unmarshal(body, &ccRequest); err != nil {
 		log.Panic(err)
 	}
+	ccRequest.InterfaceName = strings.ToLower(ccRequest.InterfaceName)
 
-	inputsName, err := s.getInputNameFromCustom(id, inputMap.NodeName, inputMap.CustomCommandName)
+	inputsName, err := s.getInputNameFromCustom(id, ccRequest.NodeName, ccRequest.InterfaceName, ccRequest.CustomCommandName)
 	if err != nil {
 		log.Panic(err)
 	}
 
 	data := make(map[string]string)
 
 	// For now custom commands are for all instances
-	instances, err := deployments.GetNodeInstancesIds(s.consulClient.KV(), id, inputMap.NodeName)
-	data[path.Join("nodes", inputMap.NodeName)] = strings.Join(instances, ",")
-	data["commandName"] = inputMap.CustomCommandName
+	instances, err := deployments.GetNodeInstancesIds(s.consulClient.KV(), id, ccRequest.NodeName)
+	data[path.Join("nodes", ccRequest.NodeName)] = strings.Join(instances, ",")
+	data["commandName"] = ccRequest.CustomCommandName
+	data["interfaceName"] = ccRequest.InterfaceName
 
 	for _, name := range inputsName {
 		if err != nil {
 			log.Panic(err)
 		}
-		data[path.Join("inputs", name)] = inputMap.Inputs[name].String()
+		data[path.Join("inputs", name)] = ccRequest.Inputs[name].String()
 	}
 
 	taskID, err := s.tasksCollector.RegisterTaskWithData(id, tasks.TaskTypeCustomCommand, data)
@@ -88,41 +91,26 @@ func (s *Server) newCustomCommandHandler(w http.ResponseWriter, r *http.Request)
 	w.WriteHeader(http.StatusAccepted)
 }
 
-func (s *Server) getInputNameFromCustom(deploymentID, nodeName, customCName string) ([]string, error) {
-	nodeType, err := deployments.GetNodeType(s.consulClient.KV(), deploymentID, nodeName)
-
+func (s *Server) getInputNameFromCustom(deploymentID, nodeName, interfaceName, customCName string) ([]string, error) {
+	kv := s.consulClient.KV()
+	op, err := operations.GetOperation(context.Background(), kv, deploymentID, nodeName, interfaceName+"."+customCName, "", "")
 	if err != nil {
 		return nil, err
 	}
-
-	kv := s.consulClient.KV()
-	kvp, _, err := kv.Keys(path.Join(consulutil.DeploymentKVPrefix, deploymentID, "topology/types", nodeType, "interfaces/custom", customCName, "inputs")+"/", "/", nil)
+	inputs, err := deployments.GetOperationInputs(kv, deploymentID, op.ImplementedInNodeTemplate, op.ImplementedInType, op.Name)
 	if err != nil {
 		log.Panic(err)
 	}
 
-	var result []string
-
-	for _, key := range kvp {
-		res, _, err := kv.Get(path.Join(key, "is_property_definition"), nil)
-
+	result := inputs[:0]
+	for _, inputName := range inputs {
+		isPropDef, err := deployments.IsOperationInputAPropertyDefinition(kv, deploymentID, op.ImplementedInNodeTemplate, op.ImplementedInType, op.Name, inputName)
 		if err != nil {
 			return nil, err
 		}
-
-		if res == nil {
-			continue
-		}
-
-		isPropDef, err := strconv.ParseBool(string(res.Value))
-		if err != nil {
-			return nil, err
-		}
-
 		if isPropDef {
-			result = append(result, path.Base(key))
+			result = append(result, inputName)
 		}
 	}
-
 	return result, nil
 }

rest/http_api.md

@@ -581,13 +581,16 @@ Request body:
 {
     "node": "NodeName",
     "name": "Custom_Command_Name",
+    "interface": "fully.qualified.interface.name",
     "inputs": {
       "index":"",
       "nb_replicas":"2"
     }
 }
 ```
 
+If omitted interface defaults to `custom`.
+
 **Response**:
 
 ```HTTP

rest/structs.go

@@ -168,6 +168,7 @@ type Attribute struct {
 type CustomCommandRequest struct {
 	NodeName          string                            `json:"node"`
 	CustomCommandName string                            `json:"name"`
+	InterfaceName     string                            `json:"interface,omitempty"`
 	Inputs            map[string]*tosca.ValueAssignment `json:"inputs"`
 }
 

tasks/workflow/task_execution.go

@@ -93,6 +93,7 @@ func (t *taskExecution) setTaskStatus(ctx context.Context, status tasks.TaskStat
 		log.Debugf("[WARNING] Failed to set task status to:%q for taskID:%q as last index has been changed before. Retry it", status.String(), t.taskID)
 		return t.checkAndSetTaskStatus(ctx, status)
 	}
+	tasks.EmitTaskEventWithContextualLogs(ctx, t.kv, t.targetID, t.taskID, t.taskType, status.String())
 	if status == tasks.TaskStatusFAILED {
 		return t.addTaskErrorFlag(ctx)
 	}

tasks/workflow/worker.go

@@ -342,6 +342,12 @@ func (w *worker) runCustomCommand(ctx context.Context, t *taskExecution) {
 		t.checkAndSetTaskStatus(ctx, tasks.TaskStatusFAILED)
 		return
 	}
+	interfaceNameKv, _, err := kv.Get(path.Join(consulutil.TasksPrefix, t.taskID, "interfaceName"), nil)
+	if err != nil {
+		log.Printf("Deployment id: %q, Task id: %q, Failed to get Custom command name: %+v", t.targetID, t.taskID, err)
+		t.checkAndSetTaskStatus(ctx, tasks.TaskStatusFAILED)
+		return
+	}
 	nodes, err := tasks.GetTaskRelatedNodes(kv, t.taskID)
 	if err != nil {
 		log.Printf("Deployment id: %q, Task id: %q, Failed to get Custom command node: %+v", t.targetID, t.taskID, err)
@@ -354,14 +360,18 @@ func (w *worker) runCustomCommand(ctx context.Context, t *taskExecution) {
 		return
 	}
 	nodeName := nodes[0]
+	interfaceName := "custom"
+	if interfaceNameKv != nil && len(interfaceNameKv.Value) != 0 {
+		interfaceName = string(interfaceNameKv.Value)
+	}
 	commandName := string(commandNameKv.Value)
 	nodeType, err := deployments.GetNodeType(w.consulClient.KV(), t.targetID, nodeName)
 	if err != nil {
 		log.Printf("Deployment id: %q, Task id: %q, Failed to get Custom command node type: %+v", t.targetID, t.taskID, err)
 		t.checkAndSetTaskStatus(ctx, tasks.TaskStatusFAILED)
 		return
 	}
-	op, err := operations.GetOperation(ctx, kv, t.targetID, nodeName, "custom."+commandName, "", "")
+	op, err := operations.GetOperation(ctx, kv, t.targetID, nodeName, interfaceName+"."+commandName, "", "")
 	if err != nil {
 		log.Printf("Deployment id: %q, Task id: %q, Command TaskExecution failed for node %q: %+v", t.targetID, t.taskID, nodeName, err)
 		err = setNodeStatus(ctx, t.kv, t.taskID, t.targetID, nodeName, tosca.NodeStateError.String())