* fix RDS/EC instance discovery for the case with more than 100 insta…

…nces

* tag-based filters for RDS/EC instances

elasticache/discovery.go

@@ -56,16 +56,43 @@ func (d *Discoverer) refresh(api *elasticache.ElastiCache) error {
 	var clusters []*elasticache.CacheCluster
 	var err error
 
-	input := &elasticache.DescribeCacheClustersInput{}
-	input.ShowCacheNodeInfo = aws.Bool(true)
-
 	for _, v := range []bool{false, true} {
+		input := &elasticache.DescribeCacheClustersInput{}
+		input.ShowCacheNodeInfo = aws.Bool(true)
 		input.ShowCacheClustersNotInReplicationGroups = aws.Bool(v)
-		output, err := api.DescribeCacheClusters(input)
-		if err != nil {
-			return err
+		for {
+			output, err := api.DescribeCacheClusters(input)
+			if err != nil {
+				return err
+			}
+			for _, cluster := range output.CacheClusters {
+				i := &elasticache.ListTagsForResourceInput{ResourceName: cluster.ARN}
+				tags := map[string]string{}
+				o, err := api.ListTagsForResource(i)
+				if err != nil {
+					d.logger.Error(err)
+				} else {
+					for _, t := range o.TagList {
+						tags[aws.StringValue(t.Key)] = aws.StringValue(t.Value)
+					}
+				}
+				if utils.Filtered(*flags.ElasticacheFilters, tags) {
+					d.logger.Infof(
+						"cluster %s (tags: %s) was skipped according to the tag-based filters: %s",
+						aws.StringValue(cluster.CacheClusterId),
+						tags,
+						*flags.ElasticacheFilters,
+					)
+					continue
+				}
+				clusters = append(clusters, cluster)
+			}
+			if output.Marker != nil {
+				input.SetMarker(aws.StringValue(output.Marker))
+				continue
+			}
+			break
 		}
-		clusters = append(clusters, output.CacheClusters...)
 	}
 
 	actualInstances := map[string]bool{}

flags/flags.go

@@ -12,5 +12,7 @@ var (
 	RdsLogsScrapeInterval     = kingpin.Flag("rds-logs-scrape-interval", "RDS logs scrape interval (0 to disable)").Default("30s").Duration()
 	DbScrapeInterval          = kingpin.Flag("db-scrape-interval", "How often to scrape DB system views").Default("30s").Duration()
 	ElasticacheConnectTimeout = kingpin.Flag("ec-connect-timeout", "Elasticache connect timeout").Default("1s").Duration()
+	ElasticacheFilters        = kingpin.Flag("ec-filter", `a tag_name:tag_value pair for filtering EC instances by their tags while discovery (env: EC_FILTER)`).Envar("EC_FILTER").StringMap()
+	RdsFilters                = kingpin.Flag("rds-filter", `a tag_name:tag_value pair for filtering RDS instances by their tags while discovery (env: RDS_FILTER)`).Envar("RDS_FILTER").StringMap()
 	ListenAddress             = kingpin.Flag("listen-address", `Listen address (env: LISTEN_ADDRESS) - "<ip>:<port>" or ":<port>".`).Envar("LISTEN_ADDRESS").Default("0.0.0.0:80").String()
 )

go.mod

@@ -11,7 +11,6 @@ require (
 	github.com/prometheus/client_golang v1.15.1
 	github.com/prometheus/memcached_exporter v0.13.0
 	gopkg.in/alecthomas/kingpin.v2 v2.2.6
-	k8s.io/klog/v2 v2.20.0
 )
 
 require (
@@ -37,4 +36,5 @@ require (
 	github.com/sirupsen/logrus v1.9.0 // indirect
 	golang.org/x/sys v0.8.0 // indirect
 	google.golang.org/protobuf v1.30.0 // indirect
+	k8s.io/klog/v2 v2.20.0 // indirect
 )

rds/discovery.go

@@ -64,6 +64,20 @@ func (d *Discoverer) refresh(api rdsiface.RDSAPI) error {
 			continue
 		}
 		id := aws.StringValue(dbInstance.DBInstanceIdentifier)
+		input := &rds.ListTagsForResourceInput{ResourceName: dbInstance.DBInstanceArn}
+		tags := map[string]string{}
+		o, err := api.ListTagsForResource(input)
+		if err != nil {
+			d.logger.Error(err)
+		} else {
+			for _, t := range o.TagList {
+				tags[aws.StringValue(t.Key)] = aws.StringValue(t.Value)
+			}
+		}
+		if utils.Filtered(*flags.RdsFilters, tags) {
+			d.logger.Infof("RDS instance %s (tags: %s) was skipped according to the tag-based filters: %s", id, tags, *flags.RdsFilters)
+			continue
+		}
 		actualInstances[id] = true
 		i, ok := d.instances[id]
 		if !ok {

utils/utils.go

@@ -3,6 +3,7 @@ package utils
 import (
 	"github.com/aws/aws-sdk-go/aws/arn"
 	"github.com/prometheus/client_golang/prometheus"
+	"path/filepath"
 	"strings"
 )
 
@@ -22,6 +23,16 @@ func IdWithRegion(region, id string) string {
 	return region + "/" + id
 }
 
+func Filtered(filters, tags map[string]string) bool {
+	for tagName, desiredValue := range filters {
+		value := tags[tagName]
+		if matched, _ := filepath.Match(desiredValue, value); !matched {
+			return true
+		}
+	}
+	return false
+}
+
 func Desc(name, help string, labels ...string) *prometheus.Desc {
 	return prometheus.NewDesc(name, help, labels, nil)
 }