Refactor deployment configuration for Kubernetes application (#558)

**What this PR does / why we need it**:

**Which issue(s) this PR fixes**:

Fixes #

**Does this PR introduce a user-facing change?**:
<!--
If no, just write "NONE" in the release-note block below.
-->
```release-note
NONE
```

This PR was merged by Kapetanios.

examples/kubernetes/bluegreen/.pipe.yaml

@@ -25,6 +25,3 @@ spec:
           primary: 100
       # Destroy all workloads of CANARY variant.
       - name: K8S_CANARY_CLEAN
-  # This example is not using service mesh.
-  trafficSplit:
-    method: pod

pkg/app/piped/executor/kubernetes/baseline.go

@@ -137,43 +137,33 @@ func (e *Executor) removeBaselineResources(ctx context.Context, resources []stri
 }
 
 func (e *Executor) generateBaselineManifests(manifests []provider.Manifest, opts config.K8sBaselineRolloutStageOptions) ([]provider.Manifest, error) {
-	var (
-		workloadKind, workloadName string
-		serviceName                string
-		generateService            bool
-		baselineManifests          []provider.Manifest
-		suffix                     = baselineVariant
-	)
-
-	// Apply the specified configuration if they are present.
-	if sc := e.config.BaselineVariant; sc != nil {
-		var ok bool
-		if sc.Suffix != "" {
-			suffix = sc.Suffix
-		}
-		generateService = sc.Service.Create
-
-		workloadKind, workloadName, ok = config.ParseVariantResourceReference(sc.Workload.Reference)
-		if !ok {
-			return nil, fmt.Errorf("malformed workload reference: %s", sc.Workload.Reference)
+	suffix := baselineVariant
+	if opts.Suffix != "" {
+		suffix = opts.Suffix
+	}
+
+	var workloads []provider.Manifest
+	if len(e.config.Workloads) == 0 {
+		workloads = findManifests(provider.KindDeployment, "", manifests)
+	} else {
+		for _, ref := range e.config.Workloads {
+			kind := provider.KindDeployment
+			if ref.Kind != "" {
+				kind = ref.Kind
+			}
+			ms := findManifests(kind, ref.Name, manifests)
+			workloads = append(workloads, ms...)
 		}
-
-		_, serviceName, ok = config.ParseVariantResourceReference(sc.Service.Reference)
-		if !ok {
-			return nil, fmt.Errorf("malformed service reference: %s", sc.Service.Reference)
-		}
-	}
-	if workloadKind == "" {
-		workloadKind = provider.KindDeployment
 	}
-
-	workloads := findManifests(workloadKind, workloadName, manifests)
 	if len(workloads) == 0 {
 		return nil, fmt.Errorf("unable to find any workload manifests for BASELINE variant")
 	}
 
+	var baselineManifests []provider.Manifest
+
 	// Find service manifests and duplicate them for BASELINE variant.
-	if generateService {
+	if opts.CreateService {
+		serviceName := e.config.Service.Name
 		services := findManifests(provider.KindService, serviceName, manifests)
 		if len(services) == 0 {
 			return nil, fmt.Errorf("unable to find any service for name=%q", serviceName)

pkg/app/piped/executor/kubernetes/canary.go

@@ -138,43 +138,33 @@ func (e *Executor) removeCanaryResources(ctx context.Context, resources []string
 }
 
 func (e *Executor) generateCanaryManifests(manifests []provider.Manifest, opts config.K8sCanaryRolloutStageOptions) ([]provider.Manifest, error) {
-	var (
-		workloadKind, workloadName string
-		serviceName                string
-		generateService            bool
-		canaryManifests            []provider.Manifest
-		suffix                     = canaryVariant
-	)
-
-	// Apply the specified configuration if they are present.
-	if sc := e.config.CanaryVariant; sc != nil {
-		var ok bool
-		if sc.Suffix != "" {
-			suffix = sc.Suffix
-		}
-		generateService = sc.Service.Create
-
-		workloadKind, workloadName, ok = config.ParseVariantResourceReference(sc.Workload.Reference)
-		if !ok {
-			return nil, fmt.Errorf("malformed workload reference: %s", sc.Workload.Reference)
+	suffix := canaryVariant
+	if opts.Suffix != "" {
+		suffix = opts.Suffix
+	}
+
+	var workloads []provider.Manifest
+	if len(e.config.Workloads) == 0 {
+		workloads = findManifests(provider.KindDeployment, "", manifests)
+	} else {
+		for _, ref := range e.config.Workloads {
+			kind := provider.KindDeployment
+			if ref.Kind != "" {
+				kind = ref.Kind
+			}
+			ms := findManifests(kind, ref.Name, manifests)
+			workloads = append(workloads, ms...)
 		}
-
-		_, serviceName, ok = config.ParseVariantResourceReference(sc.Service.Reference)
-		if !ok {
-			return nil, fmt.Errorf("malformed service reference: %s", sc.Service.Reference)
-		}
-	}
-	if workloadKind == "" {
-		workloadKind = provider.KindDeployment
 	}
-
-	workloads := findManifests(workloadKind, workloadName, manifests)
 	if len(workloads) == 0 {
 		return nil, fmt.Errorf("unable to find any workload manifests for CANARY variant")
 	}
 
+	var canaryManifests []provider.Manifest
+
 	// Find service manifests and duplicate them for CANARY variant.
-	if generateService {
+	if opts.CreateService {
+		serviceName := e.config.Service.Name
 		services := findManifests(provider.KindService, serviceName, manifests)
 		if len(services) == 0 {
 			return nil, fmt.Errorf("unable to find any service for name=%q", serviceName)

pkg/app/piped/executor/kubernetes/primary.go

@@ -48,7 +48,7 @@ func (e *Executor) ensurePrimaryRollout(ctx context.Context) model.StageStatus {
 	e.LogPersister.AppendSuccessf("Successfully loaded %d manifests", len(manifests))
 
 	// Find traffic routing manifests and filter out it from primary manifests.
-	trafficRoutingManifests, err := e.findTrafficRoutingManifests(manifests, e.config.TrafficRouting)
+	trafficRoutingManifests, err := findTrafficRoutingManifests(manifests, e.config.Service.Name, e.config.TrafficRouting)
 	if err != nil {
 		e.LogPersister.AppendErrorf("Failed while finding traffic routing manifest: (%v)", err)
 		return model.StageStatus_STAGE_FAILURE
@@ -77,7 +77,7 @@ func (e *Executor) ensurePrimaryRollout(ctx context.Context) model.StageStatus {
 
 	// Generate the manifests for applying.
 	e.LogPersister.AppendInfo("Start generating manifests for PRIMARY variant")
-	applyManifests, err := e.generatePrimaryManifests(primaryManifests)
+	applyManifests, err := e.generatePrimaryManifests(primaryManifests, *options)
 	if err != nil {
 		e.LogPersister.AppendErrorf("Unable to generate manifests for PRIMARY variant (%v)", err)
 		return model.StageStatus_STAGE_FAILURE
@@ -155,33 +155,19 @@ func findRemoveManifests(prevs []provider.Manifest, curs []provider.Manifest, na
 	return removeKeys
 }
 
-func (e *Executor) generatePrimaryManifests(manifests []provider.Manifest) ([]provider.Manifest, error) {
-	var (
-		serviceName     string
-		generateService bool
-		suffix          = primaryVariant
-	)
-
-	// Apply the specified configuration if they are present.
-	if sc := e.config.PrimaryVariant; sc != nil {
-		var ok bool
-		if sc.Suffix != "" {
-			suffix = sc.Suffix
-		}
-		generateService = sc.Service.Create
-
-		_, serviceName, ok = config.ParseVariantResourceReference(sc.Service.Reference)
-		if !ok {
-			return nil, fmt.Errorf("malformed service reference: %s", sc.Service.Reference)
-		}
+func (e *Executor) generatePrimaryManifests(manifests []provider.Manifest, opts config.K8sPrimaryRolloutStageOptions) ([]provider.Manifest, error) {
+	suffix := primaryVariant
+	if opts.Suffix != "" {
+		suffix = opts.Suffix
 	}
 
 	// Because the loaded maninests are read-only
 	// we duplicate them to avoid updating the shared manifests data in cache.
 	manifests = duplicateManifests(manifests, "")
 
 	// Find service manifests and duplicate them for PRIMARY variant.
-	if generateService {
+	if opts.CreateService {
+		serviceName := e.config.Service.Name
 		services := findManifests(provider.KindService, serviceName, manifests)
 		if len(services) == 0 {
 			return nil, fmt.Errorf("unable to find any service for name=%q", serviceName)

pkg/app/piped/executor/kubernetes/traffic.go

@@ -45,6 +45,7 @@ func (e *Executor) ensureTrafficRouting(ctx context.Context) model.StageStatus {
 		e.LogPersister.AppendErrorf("Malformed configuration for stage %s", e.Stage.Name)
 		return model.StageStatus_STAGE_FAILURE
 	}
+	method := config.DetermineTrafficRoutingMethod(e.config.TrafficRouting)
 
 	// Load the manifests at the triggered commit.
 	e.LogPersister.AppendInfof("Loading manifests at commit %s for handling", commitHash)
@@ -65,7 +66,7 @@ func (e *Executor) ensureTrafficRouting(ctx context.Context) model.StageStatus {
 	e.saveTrafficRoutingMetadata(ctx, primaryPercent, canaryPercent, baselinePercent)
 
 	// Find traffic routing manifests.
-	trafficRoutingManifests, err := e.findTrafficRoutingManifests(manifests, e.config.TrafficRouting)
+	trafficRoutingManifests, err := findTrafficRoutingManifests(manifests, e.config.Service.Name, e.config.TrafficRouting)
 	if err != nil {
 		e.LogPersister.AppendErrorf("Failed while finding traffic routing manifest: (%v)", err)
 		return model.StageStatus_STAGE_FAILURE
@@ -86,8 +87,8 @@ func (e *Executor) ensureTrafficRouting(ctx context.Context) model.StageStatus {
 	}
 	trafficRoutingManifest := trafficRoutingManifests[0]
 
-	// In case we are routing by Pod, the service manifest must contain variantLabel inside its selector.
-	if e.config.TrafficRouting == nil || e.config.TrafficRouting.Method == config.TrafficRoutingMethodPod || e.config.TrafficRouting.Method == "" {
+	// In case we are routing by PodSelector, the service manifest must contain variantLabel inside its selector.
+	if method == config.TrafficRoutingMethodPodSelector {
 		if err := checkVariantSelectorInService(trafficRoutingManifest, primaryVariant); err != nil {
 			e.LogPersister.AppendErrorf("Traffic routing by Pod requires %q inside the selector of Service manifest but it was unable to check that field in manifest %s (%v)",
 				variantLabel+": "+primaryVariant,
@@ -126,26 +127,17 @@ func (e *Executor) ensureTrafficRouting(ctx context.Context) model.StageStatus {
 	return model.StageStatus_STAGE_SUCCESS
 }
 
-func (e *Executor) findTrafficRoutingManifests(manifests []provider.Manifest, cfg *config.TrafficRouting) ([]provider.Manifest, error) {
-	if cfg != nil && cfg.Method == config.TrafficRoutingMethodIstio {
+func findTrafficRoutingManifests(manifests []provider.Manifest, serviceName string, cfg *config.TrafficRouting) ([]provider.Manifest, error) {
+	method := config.DetermineTrafficRoutingMethod(cfg)
+
+	if method == config.TrafficRoutingMethodIstio {
 		istioConfig := cfg.Istio
 		if istioConfig == nil {
 			istioConfig = &config.IstioTrafficRouting{}
 		}
 		return findIstioVirtualServiceManifests(manifests, istioConfig.VirtualService)
 	}
 
-	var podConfig config.PodTrafficRouting
-	if cfg != nil && cfg.Pod != nil {
-		podConfig = *cfg.Pod
-	}
-
-	// Find out the service which be updated the selector.
-	_, serviceName, ok := config.ParseVariantResourceReference(podConfig.Service.Reference)
-	if !ok {
-		return nil, fmt.Errorf("malformed Service reference %q", podConfig.Service.Reference)
-	}
-
 	return findManifests(provider.KindService, serviceName, manifests), nil
 }
 
@@ -195,14 +187,14 @@ func (e *Executor) saveTrafficRoutingMetadata(ctx context.Context, primary, cana
 	}
 }
 
-func findIstioVirtualServiceManifests(manifests []provider.Manifest, cfg config.K8sResourceReference) ([]provider.Manifest, error) {
+func findIstioVirtualServiceManifests(manifests []provider.Manifest, ref config.K8sResourceReference) ([]provider.Manifest, error) {
 	const (
 		istioNetworkingAPIVersionPrefix = "networking.istio.io/"
 		istioVirtualServiceKind         = "VirtualService"
 	)
-	_, name, ok := config.ParseVariantResourceReference(cfg.Reference)
-	if !ok {
-		return nil, fmt.Errorf("malformed VirtualService reference: %s", cfg.Reference)
+
+	if ref.Kind != "" && ref.Kind != istioVirtualServiceKind {
+		return nil, fmt.Errorf("support only %q kind for VirtualService reference", istioVirtualServiceKind)
 	}
 
 	var out []provider.Manifest
@@ -213,7 +205,7 @@ func findIstioVirtualServiceManifests(manifests []provider.Manifest, cfg config.
 		if m.Key.Kind != istioVirtualServiceKind {
 			continue
 		}
-		if name != "" && m.Key.Name != name {
+		if ref.Name != "" && m.Key.Name != ref.Name {
 			continue
 		}
 		out = append(out, m)