add labels flag to create cluster, refactor lables templating

Author: Jeff McCormick

client/cmd/cluster.go

@@ -26,6 +26,7 @@ import (
 	"k8s.io/client-go/pkg/api/v1"
 	"k8s.io/client-go/pkg/apis/extensions/v1beta1"
 	"strconv"
+	"strings"
 	"time"
 )
 
@@ -210,6 +211,7 @@ func createCluster(args []string) {
 
 			// Create an instance of our TPR
 			newInstance := getClusterParams(clusterName)
+			validateConfigPolicies()
 
 			newInstance.Spec.PSW_LAST_UPDATE = time.Now()
 
@@ -277,6 +279,7 @@ func getClusterParams(name string) *tpr.PgCluster {
 	spec.REPLICAS = "0"
 	spec.STRATEGY = "1"
 	spec.NodeName = NodeName
+	spec.UserLabels = UserLabelsMap
 
 	//override any values from config file
 	str := viper.GetString("CLUSTER.PORT")
@@ -394,6 +397,24 @@ func getValidNodeName() string {
 	return "error here"
 
 }
+func validateUserLabels() error {
+
+	var err error
+	labels := strings.Split(UserLabels, ",")
+
+	for _, v := range labels {
+		fmt.Printf("%s\n", v)
+		p := strings.Split(v, "=")
+		if len(p) < 2 {
+			return errors.New("invalid labels format")
+		} else {
+			UserLabelsMap[p[0]] = p[1]
+		}
+	}
+	return err
+
+}
+
 func validateNodeName(nodeName string) error {
 	var err error
 	lo := meta_v1.ListOptions{}

client/cmd/create.go

@@ -26,6 +26,8 @@ var Password string
 var SecretFrom, BackupPath, BackupPVC string
 var PoliciesFlag, PolicyFile, PolicyURL string
 var NodeName string
+var UserLabels string
+var UserLabelsMap map[string]string
 var Series int
 
 // CreateCmd represents the create command
@@ -80,6 +82,14 @@ pgo create cluster mycluster`,
 
 		}
 
+		if UserLabels != "" {
+			err = validateUserLabels()
+			if err != nil {
+				log.Error("invalid user labels, check --labels value")
+				return
+			}
+		}
+
 		if len(args) == 0 {
 			log.Error("a cluster name is required for this command")
 		} else {
@@ -124,11 +134,13 @@ func init() {
 	createClusterCmd.Flags().StringVarP(&Password, "password", "w", "", "The password to use for initial database users")
 	createClusterCmd.Flags().StringVarP(&SecretFrom, "secret-from", "s", "", "The cluster name to use when restoring secrets")
 	createClusterCmd.Flags().StringVarP(&BackupPVC, "backup-pvc", "p", "", "The backup archive PVC to restore from")
+	createClusterCmd.Flags().StringVarP(&UserLabels, "labels", "l", "", "The labels to apply to this cluster")
 	createClusterCmd.Flags().StringVarP(&BackupPath, "backup-path", "x", "", "The backup archive path to restore from")
 	createClusterCmd.Flags().StringVarP(&PoliciesFlag, "policies", "z", "", "The policies to apply when creating a cluster, comma separated")
 	createClusterCmd.Flags().StringVarP(&CCP_IMAGE_TAG, "ccp-image-tag", "c", "", "The CCP_IMAGE_TAG to use for cluster creation, if specified overrides the .pgo.yaml setting")
 	createClusterCmd.Flags().IntVarP(&Series, "series", "e", 1, "The number of clusters to create in a series, defaults to 1")
 	createPolicyCmd.Flags().StringVarP(&PolicyURL, "url", "u", "", "The url to use for adding a policy")
 	createPolicyCmd.Flags().StringVarP(&PolicyFile, "in-file", "i", "", "The policy file path to use for adding a policy")
+	UserLabelsMap = make(map[string]string)
 
 }

client/cmd/label.go

@@ -0,0 +1,256 @@
+/*
+ Copyright 2017 Crunchy Data Solutions, Inc.
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+package cmd
+
+import (
+	"fmt"
+	log "github.com/Sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/rest"
+	//"github.com/crunchydata/postgres-operator/operator/util"
+	"encoding/json"
+	"github.com/crunchydata/postgres-operator/tpr"
+	jsonpatch "github.com/evanphx/json-patch"
+	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/client-go/kubernetes"
+	//"github.com/spf13/viper"
+	//"io/ioutil"
+	//kerrors "k8s.io/apimachinery/pkg/api/errors"
+	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/pkg/apis/extensions/v1beta1"
+	//"os/user"
+	"strings"
+)
+
+var LabelCmdLabel string
+var DeleteLabel bool
+
+var labelCmd = &cobra.Command{
+	Use:   "label",
+	Short: "label a set of clusters",
+	Long: `LABEL allows you to add or remove a label on a set of clusters
+For example:
+
+pgo label mycluster yourcluster --label=environment=prod 
+pgo label mycluster yourcluster --label=environment=prod  --delete-label
+pgo label --label=environment=prod --selector=name=mycluster
+pgo label --label=environment=prod --selector=status=final --dry-run
+.`,
+	Run: func(cmd *cobra.Command, args []string) {
+		log.Debug("label called")
+		if len(args) == 0 && Selector == "" {
+			log.Error("selector or list of clusters is required to label a policy")
+			return
+		}
+		if LabelCmdLabel == "" {
+			log.Error(`You must specify the label to apply.`)
+		} else {
+			labelClusters(args)
+		}
+	},
+}
+
+func init() {
+	RootCmd.AddCommand(labelCmd)
+
+	labelCmd.Flags().StringVarP(&Selector, "selector", "s", "", "The selector to use for cluster filtering ")
+	labelCmd.Flags().StringVarP(&LabelCmdLabel, "label", "l", "", "The new label to apply for any selected or specified clusters")
+	labelCmd.Flags().BoolVarP(&DryRun, "dry-run", "d", false, "--dry-run shows clusters that label would be applied to but does not actually label them")
+	labelCmd.Flags().BoolVarP(&DeleteLabel, "delete-label", "x", false, "--delete-label deletes a label from matching clusters")
+
+}
+
+func labelClusters(clusters []string) {
+	var err error
+
+	if len(clusters) == 0 && Selector == "" {
+		fmt.Println("no clusters specified")
+		return
+	}
+	//get filtered list of pgcluster tprs
+	//get a list of all clusters
+	clusterList := tpr.PgClusterList{}
+	myselector := labels.Everything()
+	if Selector != "" {
+		log.Debug("selector is " + Selector)
+		myselector, err = labels.Parse(Selector)
+		if err != nil {
+			log.Error("could not parse --selector value " + err.Error())
+			return
+		}
+
+		log.Debugf("label selector is [%v]\n", myselector)
+		err = Tprclient.Get().
+			Resource(tpr.CLUSTER_RESOURCE).
+			Namespace(Namespace).
+			LabelsSelectorParam(myselector).
+			Do().
+			Into(&clusterList)
+		if err != nil {
+			log.Error("error getting list of clusters" + err.Error())
+			return
+		}
+		if len(clusterList.Items) == 0 {
+			fmt.Println("no clusters found")
+			return
+		}
+	} else {
+		//each arg represents a cluster name or the special 'all' value
+		items := make([]tpr.PgCluster, 0)
+		for _, cluster := range clusters {
+			result := tpr.PgCluster{}
+			err := Tprclient.Get().
+				Resource(tpr.CLUSTER_RESOURCE).
+				Namespace(Namespace).
+				Name(cluster).
+				Do().
+				Into(&result)
+			if err != nil {
+				log.Error("error getting list of clusters" + err.Error())
+				return
+			}
+			fmt.Println(result.Spec.Name)
+			items = append(items, result)
+		}
+		clusterList.Items = items
+	}
+
+	addLabels(clusterList.Items)
+
+}
+
+func addLabels(items []tpr.PgCluster) {
+	for i := 0; i < len(items); i++ {
+		fmt.Println("adding label to " + items[i].Spec.Name)
+		if DryRun {
+			fmt.Println("dry run only")
+		} else {
+			err := PatchPgCluster(Tprclient, LabelCmdLabel, items[i], Namespace)
+			if err != nil {
+				log.Error(err.Error())
+			}
+		}
+	}
+
+	for i := 0; i < len(items); i++ {
+		//get deployments for this TPR
+		lo := meta_v1.ListOptions{LabelSelector: "pg-cluster=" + items[i].Spec.Name}
+		deployments, err := Clientset.ExtensionsV1beta1().Deployments(Namespace).List(lo)
+		if err != nil {
+			log.Error("error getting list of deployments" + err.Error())
+			return
+		}
+
+		newLabels := make(map[string]string)
+		for _, d := range deployments.Items {
+			//update Deployment with the label
+			//fmt.Println(TREE_BRANCH + "deployment : " + d.ObjectMeta.Name)
+			if DryRun {
+			} else {
+				err := updateLabels(&d, Clientset, items[i].Spec.Name, Namespace, newLabels)
+				if err != nil {
+					log.Error(err.Error())
+				}
+			}
+		}
+
+	}
+}
+
+func updateLabels(deployment *v1beta1.Deployment, clientset *kubernetes.Clientset, clusterName string, namespace string, newLabels map[string]string) error {
+
+	var err error
+
+	var patchBytes, newData, origData []byte
+	origData, err = json.Marshal(deployment)
+	if err != nil {
+		return err
+	}
+
+	accessor, err2 := meta.Accessor(deployment)
+	if err2 != nil {
+		return err2
+	}
+
+	objLabels := accessor.GetLabels()
+	if objLabels == nil {
+		objLabels = make(map[string]string)
+	}
+
+	//update the deployment labels
+	for key, value := range newLabels {
+		objLabels[key] = value
+	}
+	log.Debugf("updated labels are %v\n", objLabels)
+
+	accessor.SetLabels(objLabels)
+
+	newData, err = json.Marshal(deployment)
+	if err != nil {
+		return err
+	}
+
+	patchBytes, err = jsonpatch.CreateMergePatch(origData, newData)
+	if err != nil {
+		return err
+	}
+
+	_, err = clientset.ExtensionsV1beta1().Deployments(namespace).Patch(clusterName, types.MergePatchType, patchBytes, "")
+	if err != nil {
+		log.Debug("error patching deployment " + err.Error())
+	}
+	return err
+
+}
+
+func PatchPgCluster(tprclient *rest.RESTClient, newLabel string, oldTpr tpr.PgCluster, namespace string) error {
+
+	fields := strings.Split(newLabel, "=")
+	labelKey := fields[0]
+	labelValue := fields[1]
+	oldData, err := json.Marshal(oldTpr)
+	if err != nil {
+		return err
+	}
+	if oldTpr.Metadata.Labels == nil {
+		oldTpr.Metadata.Labels = make(map[string]string)
+	}
+	oldTpr.Metadata.Labels[labelKey] = labelValue
+	var newData, patchBytes []byte
+	newData, err = json.Marshal(oldTpr)
+	if err != nil {
+		return err
+	}
+	patchBytes, err = jsonpatch.CreateMergePatch(oldData, newData)
+	if err != nil {
+		return err
+	}
+
+	log.Debug(string(patchBytes))
+
+	_, err6 := tprclient.Patch(types.MergePatchType).
+		Namespace(namespace).
+		Resource(tpr.CLUSTER_RESOURCE).
+		Name(oldTpr.Spec.Name).
+		Body(patchBytes).
+		Do().
+		Get()
+
+	return err6
+
+}