Repository: zalando-incubator/kube-ingress-aws-controller
Pull request #277 (2019-10-02 17:40:39)

Rework fix for CUSTOM_FILTERS
This PR is to address issue #266 and obsoletes PR #275.

It should not cause a regression to the patches which caused this regression. :)

Signed-off-by: Joe Hohertz <joe@viafoura.com>

coveralls: 
[![Coverage Status](https://coveralls.io/builds/26072233/badge)](https://coveralls.io/builds/26072233)

Coverage increased (+0.2%) to 47.407% when pulling **73b4153f12949c8134f61e6fc5352bdcea5514cf on viafoura:fix-custom-filters-take-3** into **bd660aa5cbeafd030572afc083c8aba21c73d81a on zalando-incubator:master**.

jhohertz: I :heart: playing please the linter... No idea what it was on about, but did a gofmt on the file, and things seem happy now.
szuecs: @jhohertz can you please create table driven unit tests for these 2 functions?

jhohertz: Will do.
jhohertz: Adjusted so the parsing bits are a bit better positioned (as the existing instance filters are handled), and rebased to current master

jhohertz: Rebased
jhohertz: Huh... missed that in my testing. Looking into it.
szuecs: @jhohertz to advertise your work and show how to use it, there is also documentation missing. Maybe a section in the README.md :)

Code and tests look good to me
jhohertz: @szuecs : So strictly speaking, this PR is simply to restore the functionality already described in the [how it works section of README.md](https://github.com/zalando-incubator/kube-ingress-aws-controller/blob/master/README.md#how-it-works) which stopped working in v0.8.4.

That said... There is now a small inaccuracy due to the fact that the autoscale filters API does not implement the same way as the EC2 instance tags. (And further because I avoided even using that API).

The new implementation can not support the entire family of filters of the EC2 API. I've kind of faked the `tag` and `tag-key` prefixes to support existing uses of same. However tags like `vpc` and the rest of the EC2 API are just not available. My recommendation is to change the text:

```
where name is one of names that are recognized by the EC2 API (you can find list here)
```

to

```
where name can be a tag: or tag-key: prefix as would be recognised by the EC2 API
```
I did consider also the possibility of constructing a separate "tag cloud" for comparison, from the actual tags, and additional ones that would include `vpc` and anything else that is both available from the ASG description, AND is listed on the EC2 API, and so could be remapped in this manner. Then we could list in the docs any "ec2-like" filters that are supported.

That said... I seem to be the only one that this feature regression seemed to bother, and we aren't using anything besides "tag" and "tag-key" type filters... so.... maybe it's not worth adding the additional complexity here.

Interested in your, and others, thoughts on this before we finalise it.


szuecs: From my side you can do the minimal change that you suggested.
jhohertz: Made the change, dropped mention of the 'vpc' tag since it's not supported, and expanded the examples a bit.
szuecs: :+1: 
mikkeloscar: :+1: 

Patch:
diff --git a/README.md b/README.md
index 786a63a..260657b 100644
--- a/README.md
+++ b/README.md
@@ -154,12 +154,13 @@ During startup phase EC2 filters are constructed as follows:
   filters are `tag:kubernetes.io/cluster/<cluster-id>=owned tag-key=k8s.io/role/node` where `<cluster-id>`
   is determined from EC2 tags of instance on which Ingress Controller pod is started.
 
-`CUSTOM_FILTERS` is a list of filters separated by spaces. Each filter has a form of `name=value` where name is one
-of names that are recognized by the EC2 API (you can find list [here](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeInstances.html))
-and value is value of a filter. For example:
+`CUSTOM_FILTERS` is a list of filters separated by spaces. Each filter has a form of `name=value` where name can be a `tag:` or `tag-key:` prefixed expression, as would be recognized by the EC2 API, and value is value of a filter, or a comma seperated list of values. 
 
-* `tag-key=test` will filter instances that has tag named `test`.
-* `vpc-id=vpc-1a2b3c4d` will filter instances that belong to specific VPC.
+For example:
+
+* `tag-key=test` will filter instances that have a tag named `test`, ignoring the value.
+* `tag:foo=bar' will filter instances that have a tag named `foo` with the value `bar`
+* `tag:abc=def,ghi' will filter instances that have a tag named `abc` with the value `def` OR `ghi`
 * Default filter `tag:kubernetes.io/cluster/<cluster-id>=owned tag-key=k8s.io/role/node` filters instances
   that has tag `kubernetes.io/cluster/<cluster-id>` with value `owned` and have tag named `tag-key=k8s.io/role/node`.
 
diff --git a/aws/adapter.go b/aws/adapter.go
index 2767996..35a6f23 100644
--- a/aws/adapter.go
+++ b/aws/adapter.go
@@ -6,8 +6,6 @@ import (
 	"strings"
 	"time"
 
-	"os"
-
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/client"
 	"github.com/aws/aws-sdk-go/aws/ec2metadata"
@@ -60,6 +58,7 @@ type Adapter struct {
 	albLogsS3Prefix            string
 	wafWebAclId                string
 	httpRedirectToHttps        bool
+	customFilter               string
 }
 
 type manifest struct {
@@ -67,6 +66,7 @@ type manifest struct {
 	instance      *instanceDetails
 	subnets       []*subnetDetails
 	filters       []*ec2.Filter
+	asgFilters    map[string][]string
 }
 
 type configProviderFunc func() client.ConfigProvider
@@ -97,11 +97,10 @@ const (
 	// DefaultAlbS3LogsPrefix is a blank string, and optionally set if desired
 	DefaultAlbS3LogsPrefix = ""
 	DefaultWafWebAclId     = ""
+	DefaultCustomFilter    = ""
 
 	ipAddressTypeDualstack = "dualstack"
 	nameTag                = "Name"
-
-	customTagFilterEnvVarName = "CUSTOM_FILTERS"
 )
 
 var (
@@ -174,6 +173,7 @@ func NewAdapter(newControllerID string) (adapter *Adapter, err error) {
 		albLogsS3Bucket:     DefaultAlbS3LogsBucket,
 		albLogsS3Prefix:     DefaultAlbS3LogsPrefix,
 		wafWebAclId:         DefaultWafWebAclId,
+		customFilter:        DefaultCustomFilter,
 	}
 
 	adapter.manifest, err = buildManifest(adapter)
@@ -299,6 +299,15 @@ func (a *Adapter) WithHttpRedirectToHttps(httpRedirectToHttps bool) *Adapter {
 	return a
 }
 
+// WithCustomFilter returns the receiver adapter after setting a custom filter expression
+func (a *Adapter) WithCustomFilter(customFilter string) *Adapter {
+	a.customFilter = customFilter
+	// also rebuild related manifest items
+	a.manifest.filters = a.parseFilters(a.ClusterID())
+	a.manifest.asgFilters = a.parseAutoscaleFilterTags(a.ClusterID())
+	return a
+}
+
 // ClusterID returns the ClusterID tag that all resources from the same Kubernetes cluster share.
 // It's taken from the current ec2 instance.
 func (a *Adapter) ClusterID() string {
@@ -575,7 +584,8 @@ func buildManifest(awsAdapter *Adapter) (*manifest, error) {
 		securityGroup: securityGroupDetails,
 		instance:      instanceDetails,
 		subnets:       subnets,
-		filters:       parseFilters(clusterID),
+		filters:       awsAdapter.parseFilters(clusterID),
+		asgFilters:    awsAdapter.parseAutoscaleFilterTags(clusterID),
 	}, nil
 }
 
@@ -680,7 +690,7 @@ func (a *Adapter) UpdateAutoScalingGroupsAndInstances() error {
 	}
 
 	newAutoScalingGroups := make(map[string]*autoScalingGroupDetails)
-	fetchedAutoScalingGroups, err := getOwnedAutoScalingGroups(a.autoscaling, a.ClusterID())
+	fetchedAutoScalingGroups, err := getOwnedAutoScalingGroups(a.autoscaling, a.manifest.asgFilters)
 	if err != nil {
 		return err
 	}
@@ -697,14 +707,14 @@ func (a *Adapter) UpdateAutoScalingGroupsAndInstances() error {
 // later on each cycle. Filter is based on value of customTagFilterEnvVarName environment
 // veriable. If it is undefined or could not be parsed, default filter is returned which
 // filters on kubernetesClusterTag tag value and kubernetesNodeRoleTag existance.
-func parseFilters(clusterId string) []*ec2.Filter {
-	if filter, ok := os.LookupEnv(customTagFilterEnvVarName); ok {
-		terms := strings.Fields(filter)
+func (a *Adapter) parseFilters(clusterId string) []*ec2.Filter {
+	if a.customFilter != "" {
+		terms := strings.Fields(a.customFilter)
 		filters := make([]*ec2.Filter, len(terms))
 		for i, term := range terms {
 			parts := strings.Split(term, "=")
 			if len(parts) != 2 {
-				log.Errorf("failed parsing %s, falling back to default", customTagFilterEnvVarName)
+				log.Errorf("failed parsing %s, falling back to default", a.customFilter)
 				return generateDefaultFilters(clusterId)
 			}
 			filters[i] = &ec2.Filter{
@@ -731,3 +741,36 @@ func generateDefaultFilters(clusterId string) []*ec2.Filter {
 		},
 	}
 }
+
+// We look to the same string used for instance filtering, however, we are much more limited in what can be done for
+// ASGs. As such, we instead build a map of tags to look for as we iterate over all ASGs in getOwnedAutoScalingGroups
+func (a *Adapter) parseAutoscaleFilterTags(clusterId string) map[string][]string {
+	if a.customFilter != "" {
+		terms := strings.Fields(a.customFilter)
+		filterTags := make(map[string][]string)
+		for _, term := range terms {
+			parts := strings.Split(term, "=")
+			if len(parts) != 2 {
+				log.Errorf("failed parsing %s, falling back to default", a.customFilter)
+				return generateDefaultAutoscaleFilterTags(clusterId)
+			}
+			if parts[0] == "tag-key" {
+				filterTags[parts[1]] = []string{}
+			} else if strings.HasPrefix(parts[0], "tag:") {
+				tagparts := strings.Split(parts[0], ":")
+				filterTags[tagparts[1]] = strings.Split(parts[1], ",")
+			} else {
+				filterTags[parts[0]] = strings.Split(parts[1], ",")
+			}
+		}
+		return filterTags
+	}
+	return generateDefaultAutoscaleFilterTags(clusterId)
+
+}
+
+func generateDefaultAutoscaleFilterTags(clusterId string) map[string][]string {
+	filterTags := make(map[string][]string)
+	filterTags[clusterIDTagPrefix+clusterId] = []string{resourceLifecycleOwned}
+	return filterTags
+}
diff --git a/aws/asg.go b/aws/asg.go
index 688257d..4009029 100644
--- a/aws/asg.go
+++ b/aws/asg.go
@@ -85,10 +85,36 @@ func getAutoScalingGroupsByName(service autoscalingiface.AutoScalingAPI, autoSca
 	return result, nil
 }
 
-func getOwnedAutoScalingGroups(service autoscalingiface.AutoScalingAPI, clusterID string) (map[string]*autoScalingGroupDetails, error) {
-	params := &autoscaling.DescribeAutoScalingGroupsInput{}
+// Given a set of filter tags, and actual ASG tags, iterate over every filter tag,
+// looking for a matching tag name on the ASG. If one is seen, and our filter value is
+// empty, or contains the value on the ASG tag, count it as a match. If all matched,
+// Test as true, otherwise return false
+func testFilterTags(filterTags map[string][]string, asgTags map[string]string) bool {
+	matches := make(map[string]int)
+	for filterKey, filterValues := range filterTags {
+		if v, found := asgTags[filterKey]; found {
+			if len(filterValues) == 0 {
+				matches[filterKey] = matches[filterKey] + 1
+			} else {
+				for _, filterVal := range filterValues {
+					if v == filterVal {
+						matches[filterKey] = matches[filterKey] + 1
+					}
+				}
+			}
+		} else {
+			// failed to match, return fast
+			return false
+		}
+	}
+	if len(filterTags) == len(matches) {
+		return true
+	}
+	return false
+}
 
-	clusterIDTag := clusterIDTagPrefix + clusterID
+func getOwnedAutoScalingGroups(service autoscalingiface.AutoScalingAPI, filterTags map[string][]string) (map[string]*autoScalingGroupDetails, error) {
+	params := &autoscaling.DescribeAutoScalingGroupsInput{}
 
 	result := make(map[string]*autoScalingGroupDetails)
 	err := service.DescribeAutoScalingGroupsPages(params,
@@ -96,19 +122,14 @@ func getOwnedAutoScalingGroups(service autoscalingiface.AutoScalingAPI, clusterI
 			for _, g := range page.AutoScalingGroups {
 				name := aws.StringValue(g.AutoScalingGroupName)
 
-				isOwned := false
 				tags := make(map[string]string)
 				for _, td := range g.Tags {
 					key := aws.StringValue(td.Key)
 					value := aws.StringValue(td.Value)
 					tags[key] = value
-
-					if key == clusterIDTag && value == resourceLifecycleOwned {
-						isOwned = true
-					}
 				}
 
-				if isOwned {
+				if testFilterTags(filterTags, tags) {
 					result[name] = &autoScalingGroupDetails{
 						name:                    name,
 						arn:                     aws.StringValue(g.AutoScalingGroupARN),
diff --git a/controller.go b/controller.go
index 993efc7..d679ffc 100644
--- a/controller.go
+++ b/controller.go
@@ -27,6 +27,7 @@ const (
 	defaultDisableSNISupport   = false
 	defaultHttpRedirectToHttps = false
 	defaultCertTTL             = 30 * time.Minute
+	customTagFilterEnvVarName  = "CUSTOM_FILTERS"
 )
 
 var (
@@ -228,6 +229,12 @@ func main() {
 	if err != nil {
 		log.Fatal(err)
 	}
+
+	customFilter, ok := os.LookupEnv(customTagFilterEnvVarName)
+	if !ok {
+		customFilter = ""
+	}
+
 	awsAdapter = awsAdapter.
 		WithHealthCheckPath(healthCheckPath).
 		WithHealthCheckPort(healthCheckPort).
@@ -243,7 +250,8 @@ func main() {
 		WithAlbLogsS3Bucket(albLogsS3Bucket).
 		WithAlbLogsS3Prefix(albLogsS3Prefix).
 		WithWafWebAclId(wafWebAclId).
-		WithHttpRedirectToHttps(httpRedirectToHttps)
+		WithHttpRedirectToHttps(httpRedirectToHttps).
+		WithCustomFilter(customFilter)
 
 	certificatesProvider, err := certs.NewCachingProvider(
 		certPollingInterval,


Test patch:
diff --git a/aws/adapter_test.go b/aws/adapter_test.go
index d0b747c..18cce7b 100644
--- a/aws/adapter_test.go
+++ b/aws/adapter_test.go
@@ -2,7 +2,6 @@ package aws
 
 import (
 	"fmt"
-	"os"
 	"reflect"
 	"sort"
 	"testing"
@@ -147,12 +146,20 @@ func TestParseFilters(tt *testing.T) {
 	} {
 		tt.Run(fmt.Sprintf("%v", test.name), func(t *testing.T) {
 			tt.Log(test.name)
+			a := &Adapter{
+				ec2Details: map[string]*instanceDetails{},
+				manifest: &manifest{
+					instance: &instanceDetails{
+						tags: map[string]string{
+							clusterIDTagPrefix + test.clusterId: resourceLifecycleOwned,
+						},
+					},
+				},
+			}
 			if test.customFilter != nil {
-				os.Setenv(customTagFilterEnvVarName, *test.customFilter)
-			} else {
-				os.Unsetenv(customTagFilterEnvVarName)
+				a = a.WithCustomFilter(*test.customFilter)
 			}
-			output := parseFilters(test.clusterId)
+			output := a.parseFilters(test.clusterId)
 			if !reflect.DeepEqual(test.expectedFilters, output) {
 				t.Errorf("unexpected result. wanted %q, got %q", test.expectedFilters, output)
 			}
@@ -767,3 +774,95 @@ func TestFindLBSubnets(tt *testing.T) {
 		})
 	}
 }
+
+func TestParseFilterTagsDefault(t *testing.T) {
+	for _, test := range []struct {
+		name      string
+		clusterId string
+		want      map[string][]string
+	}{
+		{
+			"success-default-filter-asg",
+			"mycluster",
+			map[string][]string{
+				clusterIDTagPrefix + "mycluster": []string{resourceLifecycleOwned},
+			},
+		},
+	} {
+		t.Run(fmt.Sprintf("%v", test.name), func(t *testing.T) {
+			a := Adapter{
+				customFilter: "",
+				ec2Details:   map[string]*instanceDetails{},
+				manifest: &manifest{
+					instance: &instanceDetails{
+						tags: map[string]string{
+							clusterIDTagPrefix + test.clusterId: resourceLifecycleOwned,
+						},
+					},
+				},
+			}
+			got := a.parseAutoscaleFilterTags(test.clusterId)
+
+			if !reflect.DeepEqual(test.want, got) {
+				t.Errorf("unexpected result. wanted %+v, got %+v", test.want, got)
+			}
+		})
+	}
+}
+
+func TestParseFilterTagsCustom(t *testing.T) {
+	for _, test := range []struct {
+		name         string
+		clusterId    string
+		customFilter string
+		want         map[string][]string
+	}{
+		{
+			"success-custom-filter-asg",
+			"mycluster",
+			"tag:kubernetes.io/cluster/mycluster=owned tag-key=k8s.io/role/node tag-key=custom.com/ingress",
+			map[string][]string{
+				"kubernetes.io/cluster/mycluster": []string{"owned"},
+				"k8s.io/role/node":                []string{},
+				"custom.com/ingress":              []string{},
+			},
+		},
+		{
+			"success-custom-filter-multivalue-asg",
+			"mycluster",
+			"tag:kubernetes.io/cluster/mycluster=owned tag-key=k8s.io/role/node tag:custom.com/ingress=owned,shared",
+			map[string][]string{
+				"kubernetes.io/cluster/mycluster": []string{"owned"},
+				"k8s.io/role/node":                []string{},
+				"custom.com/ingress":              []string{"owned", "shared"},
+			},
+		},
+		{
+			"success-custom-filter-fallback-to-default-asg",
+			"mycluster",
+			"tag:goodtag=foo tag-key=alsogood thisisabadtag andthisonetoo",
+			map[string][]string{
+				clusterIDTagPrefix + "mycluster": []string{resourceLifecycleOwned},
+			},
+		},
+	} {
+		t.Run(fmt.Sprintf("%v", test.name), func(t *testing.T) {
+			a := Adapter{
+				customFilter: test.customFilter,
+				ec2Details:   map[string]*instanceDetails{},
+				manifest: &manifest{
+					instance: &instanceDetails{
+						tags: map[string]string{
+							clusterIDTagPrefix + test.clusterId: resourceLifecycleOwned,
+						},
+					},
+				},
+			}
+			got := a.parseAutoscaleFilterTags(test.clusterId)
+
+			if !reflect.DeepEqual(test.want, got) {
+				t.Errorf("unexpected result. wanted %+v, got %+v", test.want, got)
+			}
+		})
+	}
+}
diff --git a/aws/asg_test.go b/aws/asg_test.go
index 7268948..0e0daf1 100644
--- a/aws/asg_test.go
+++ b/aws/asg_test.go
@@ -475,3 +475,57 @@ func TestDetach(t *testing.T) {
 	}
 
 }
+
+func TestTestFilterTags(t *testing.T) {
+	for _, test := range []struct {
+		name         string
+		clusterId    string
+		customFilter string
+		asgTags      map[string]string
+		want         bool
+	}{
+		{
+			"success-test-filter-asg",
+			"mycluster",
+			"tag:kubernetes.io/cluster/mycluster=owned tag-key=k8s.io/role/node tag-key=custom.com/ingress",
+			map[string]string{
+				"kubernetes.io/cluster/mycluster": "owned",
+				"k8s.io/role/node":                "1",
+				"custom.com/ingress":              "",
+			},
+			true,
+		},
+		{
+			"success-test-filter-negative-asg",
+			"mycluster",
+			"tag:kubernetes.io/cluster/yourcluster=owned tag-key=k8s.io/role/node tag-key=custom.com/ingress",
+			map[string]string{
+				"kubernetes.io/cluster/mycluster": "owned",
+				"k8s.io/role/node":                "1",
+				"custom.com/ingress":              "",
+			},
+			false,
+		},
+		{
+			"success-test-filter-multi-value-asg",
+			"mycluster",
+			"tag:kubernetes.io/cluster/mycluster=owned tag-key=k8s.io/role/node tag:custom.com/ingress=owned,shared",
+			map[string]string{
+				"kubernetes.io/cluster/mycluster": "owned",
+				"k8s.io/role/node":                "1",
+				"custom.com/ingress":              "shared",
+			},
+			true,
+		},
+	} {
+		t.Run(fmt.Sprintf("%v", test.name), func(t *testing.T) {
+			a := &Adapter{customFilter: test.customFilter}
+			filterTags := a.parseAutoscaleFilterTags(test.clusterId)
+			got := testFilterTags(filterTags, test.asgTags)
+
+			if !reflect.DeepEqual(test.want, got) {
+				t.Errorf("unexpected result. wanted %+v, got %+v", test.want, got)
+			}
+		})
+	}
+}