Tag cortex resources (#1031)

cli/cmd/cluster.go

@@ -142,11 +142,19 @@ var _upCmd = &cobra.Command{
 		if err != nil {
 			exit.Error(err)
 		}
+		err = awsClient.TagBucket(clusterConfig.Bucket, clusterConfig.Tags)
+		if err != nil {
+			exit.Error(err)
+		}
 
 		err = CreateLogGroupIfNotFound(awsClient, clusterConfig.LogGroup)
 		if err != nil {
 			exit.Error(err)
 		}
+		err = awsClient.TagLogGroup(clusterConfig.LogGroup, clusterConfig.Tags)
+		if err != nil {
+			exit.Error(err)
+		}
 
 		out, exitCode, err := runManagerUpdateCommand("/root/install.sh", clusterConfig, awsCreds, _flagClusterEnv)
 		if err != nil {

cli/cmd/lib_cluster_config.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"path"
 	"path/filepath"
+	"reflect"
 	"regexp"
 
 	"github.com/cortexlabs/cortex/pkg/consts"
@@ -229,6 +230,10 @@ func getClusterConfigureConfig(cachedClusterConfig clusterconfig.Config, awsCred
 		}
 		userClusterConfig.InstanceType = cachedClusterConfig.InstanceType
 
+		if !reflect.DeepEqual(userClusterConfig.Tags, cachedClusterConfig.Tags) {
+			return nil, clusterconfig.ErrorConfigCannotBeChangedOnUpdate(clusterconfig.TagsKey, s.ObjFlat(*&cachedClusterConfig.Tags))
+		}
+
 		if len(userClusterConfig.AvailabilityZones) > 0 && !strset.New(userClusterConfig.AvailabilityZones...).IsEqual(strset.New(cachedClusterConfig.AvailabilityZones...)) {
 			return nil, clusterconfig.ErrorConfigCannotBeChangedOnUpdate(clusterconfig.AvailabilityZonesKey, cachedClusterConfig.AvailabilityZones)
 		}
@@ -486,6 +491,7 @@ func clusterConfigConfirmaionStr(clusterConfig clusterconfig.Config, awsCreds AW
 	items.Add(clusterconfig.InstanceTypeUserKey, *clusterConfig.InstanceType)
 	items.Add(clusterconfig.MinInstancesUserKey, *clusterConfig.MinInstances)
 	items.Add(clusterconfig.MaxInstancesUserKey, *clusterConfig.MaxInstances)
+	items.Add(clusterconfig.TagsKey, s.ObjFlatNoQuotes(clusterConfig.Tags))
 	if clusterConfig.InstanceVolumeSize != defaultConfig.InstanceVolumeSize {
 		items.Add(clusterconfig.InstanceVolumeSizeUserKey, clusterConfig.InstanceVolumeSize)
 	}

docs/cluster-management/config.md

@@ -65,6 +65,9 @@ operator_load_balancer_scheme: internet-facing  # must be "internet-facing" or "
 # CloudWatch log group for cortex (default: <cluster_name>)
 log_group: cortex
 
+# additional tags to assign to aws resources for labelling and cost allocation (by default, all resources will be tagged with cortex.dev/cluster-name=<cluster_name>)
+tags:  # <string>: <string> map of key/value pairs
+
 # whether to use spot instances in the cluster (default: false)
 # see https://cortex.dev/v/master/cluster-management/spot-instances for additional details on spot configuration
 spot: false

manager/cluster_config_env.py

@@ -17,6 +17,11 @@
 
 
 def export(base_key, value):
+    if base_key.lower().startswith("cortex_tags"):
+        inlined_tags = ",".join([f"{k}={v}" for k, v in value.items()])
+        print(f"export CORTEX_TAGS={inlined_tags}")
+        return
+
     if value is None:
         return
     elif type(value) is list:

manager/generate_eks.py

@@ -153,6 +153,7 @@ def generate_eks(cluster_config_path):
             "name": cluster_config["cluster_name"],
             "region": cluster_config["region"],
             "version": "1.15",
+            "tags": cluster_config["tags"],
         },
         "vpc": {"nat": {"gateway": nat_gateway}},
         "availabilityZones": cluster_config["availability_zones"],

manager/manifests/istio-values.yaml

@@ -41,6 +41,7 @@ gateways:
     serviceAnnotations:
       ${CORTEX_OPERATOR_LOAD_BALANCER_ANNOTATION}
       service.beta.kubernetes.io/aws-load-balancer-type: "nlb"
+      service.beta.kubernetes.io/aws-load-balancer-additional-resource-tags: ${CORTEX_TAGS}
     type: LoadBalancer
     externalTrafficPolicy: Local  # https://medium.com/pablo-perez/k8s-externaltrafficpolicy-local-or-cluster-40b259a19404, https://www.asykim.com/blog/deep-dive-into-kubernetes-external-traffic-policies
     ports:
@@ -83,6 +84,7 @@ gateways:
     serviceAnnotations:
       ${CORTEX_API_LOAD_BALANCER_ANNOTATION}
       service.beta.kubernetes.io/aws-load-balancer-type: "nlb"
+      service.beta.kubernetes.io/aws-load-balancer-additional-resource-tags: ${CORTEX_TAGS}
     type: LoadBalancer
     externalTrafficPolicy: Local  # https://medium.com/pablo-perez/k8s-externaltrafficpolicy-local-or-cluster-40b259a19404, https://www.asykim.com/blog/deep-dive-into-kubernetes-external-traffic-policies
     ports:

pkg/lib/aws/cloudwatch.go

@@ -46,3 +46,21 @@ func (c *Client) CreateLogGroup(logGroup string) error {
 
 	return nil
 }
+
+func (c *Client) TagLogGroup(logGroup string, tagMap map[string]string) error {
+	tags := map[string]*string{}
+	for key, value := range tagMap {
+		tags[key] = aws.String(value)
+	}
+
+	_, err := c.CloudWatchLogs().TagLogGroup(&cloudwatchlogs.TagLogGroupInput{
+		LogGroupName: aws.String(logGroup),
+		Tags:         tags,
+	})
+
+	if err != nil {
+		return errors.Wrap(err, "failed to add tags to log group", logGroup)
+	}
+
+	return nil
+}

pkg/lib/aws/s3.go

@@ -698,3 +698,28 @@ func (c *Client) S3BatchIterator(bucket string, prefix string, includeDirObjects
 
 	return nil
 }
+
+func (c *Client) TagBucket(bucket string, tagMap map[string]string) error {
+	var tagSet []*s3.Tag
+	for key, value := range tagMap {
+		tagSet = append(tagSet, &s3.Tag{
+			Key:   aws.String(key),
+			Value: aws.String(value),
+		})
+	}
+
+	_, err := c.S3().PutBucketTagging(
+		&s3.PutBucketTaggingInput{
+			Bucket: aws.String(bucket),
+			Tagging: &s3.Tagging{
+				TagSet: tagSet,
+			},
+		},
+	)
+
+	if err != nil {
+		return errors.Wrap(err, "failed to add tags to bucket", bucket)
+	}
+
+	return nil
+}

pkg/lib/configreader/interface_map.go

@@ -27,6 +27,7 @@ type InterfaceMapValidation struct {
 	Default                map[string]interface{}
 	AllowExplicitNull      bool
 	AllowEmpty             bool
+	ConvertNilToEmpty      bool
 	ScalarsOnly            bool
 	StringLeavesOnly       bool
 	StringKeysOnly         bool
@@ -144,5 +145,10 @@ func validateInterfaceMap(val map[string]interface{}, v *InterfaceMapValidation)
 	if v.Validator != nil {
 		return v.Validator(val)
 	}
+
+	if val == nil && v.ConvertNilToEmpty {
+		val = make(map[string]interface{})
+	}
+
 	return val, nil
 }

pkg/lib/configreader/string_map.go

@@ -26,6 +26,7 @@ type StringMapValidation struct {
 	Default                map[string]string
 	AllowExplicitNull      bool
 	AllowEmpty             bool
+	ConvertNilToEmpty      bool
 	AllowCortexResources   bool
 	RequireCortexResources bool
 	Validator              func(map[string]string) (map[string]string, error)
@@ -89,5 +90,10 @@ func validateStringMap(val map[string]string, v *StringMapValidation) (map[strin
 	if v.Validator != nil {
 		return v.Validator(val)
 	}
+
+	if val == nil && v.ConvertNilToEmpty {
+		val = make(map[string]string)
+	}
+
 	return val, nil
 }

pkg/types/clusterconfig/clusterconfig.go

@@ -38,7 +38,7 @@ import (
 var (
 	_spotInstanceDistributionLength = 2
 	_maxInstancePools               = 20
-
+	_tagName                        = "cortex.dev/cluster-name"
 	// This regex is stricter than the actual S3 rules
 	_strictS3BucketRegex = regexp.MustCompile(`^([a-z0-9])+(-[a-z0-9]+)*$`)
 )
@@ -50,6 +50,7 @@ type Config struct {
 	InstanceVolumeSize         int64              `json:"instance_volume_size" yaml:"instance_volume_size"`
 	InstanceVolumeType         VolumeType         `json:"instance_volume_type" yaml:"instance_volume_type"`
 	InstanceVolumeIOPS         *int64             `json:"instance_volume_iops" yaml:"instance_volume_iops"`
+	Tags                       map[string]string  `json:"tags" yaml:"tags"`
 	Spot                       *bool              `json:"spot" yaml:"spot"`
 	SpotConfig                 *SpotConfig        `json:"spot_config" yaml:"spot_config"`
 	ClusterName                string             `json:"cluster_name" yaml:"cluster_name"`
@@ -141,6 +142,14 @@ var UserValidation = &cr.StructValidation{
 				return VolumeTypeFromString(str), nil
 			},
 		},
+		{
+			StructField: "Tags",
+			StringMapValidation: &cr.StringMapValidation{
+				AllowExplicitNull: true,
+				AllowEmpty:        true,
+				ConvertNilToEmpty: true,
+			},
+		},
 		{
 			StructField: "InstanceVolumeIOPS",
 			Int64PtrValidation: &cr.Int64PtrValidation{
@@ -513,6 +522,10 @@ func (cc *Config) Validate(awsClient *aws.Client) error {
 		}
 	}
 
+	if _, ok := cc.Tags[_tagName]; !ok {
+		cc.Tags[_tagName] = cc.ClusterName
+	}
+
 	if err := cc.validateAvailabilityZones(awsClient); err != nil {
 		return errors.Wrap(err, AvailabilityZonesKey)
 	}
@@ -991,6 +1004,7 @@ func (cc *Config) UserTable() table.KeyValuePairs {
 	items.Add(InstanceTypeUserKey, *cc.InstanceType)
 	items.Add(MinInstancesUserKey, *cc.MinInstances)
 	items.Add(MaxInstancesUserKey, *cc.MaxInstances)
+	items.Add(TagsUserKey, s.ObjFlat(cc.Tags))
 	items.Add(InstanceVolumeSizeUserKey, cc.InstanceVolumeSize)
 	items.Add(InstanceVolumeTypeUserKey, cc.InstanceVolumeType)
 	items.Add(InstanceVolumeIOPSUserKey, cc.InstanceVolumeIOPS)

pkg/types/clusterconfig/config_key.go

@@ -20,6 +20,7 @@ const (
 	InstanceTypeKey                        = "instance_type"
 	MinInstancesKey                        = "min_instances"
 	MaxInstancesKey                        = "max_instances"
+	TagsKey                                = "tags"
 	InstanceVolumeSizeKey                  = "instance_volume_size"
 	InstanceVolumeTypeKey                  = "instance_volume_type"
 	InstanceVolumeIOPSKey                  = "instance_volume_iops"
@@ -65,6 +66,7 @@ const (
 	InstanceTypeUserKey                        = "instance type"
 	MinInstancesUserKey                        = "min instances"
 	MaxInstancesUserKey                        = "max instances"
+	TagsUserKey                                = "tags"
 	InstanceVolumeSizeUserKey                  = "instance volume size (Gi)"
 	InstanceVolumeTypeUserKey                  = "instance volume type"
 	InstanceVolumeIOPSUserKey                  = "instance volume iops"