Add configurable region support (#22)

* Add configurable region support

- Adds ability to configure region via config map
- Tidies up some hardcoding around Cloud Map pre-release obsfucation and limitations

* address pr review

README.md

@@ -17,11 +17,12 @@ data:
   access-key-id: <base64-encoded-IAM-access-key-id>
   secret-access-key: <base64-encoded-IAM-secret-access-key>
 ```
-3. Deploy the Istio Cloud Map Operator:
+3. Edit the aws-config config map in `kubernetes/deployment.yaml` to choose the AWS Cloud Map region to sync with.
+4. Deploy the Istio Cloud Map Operator:
 ```bash
-$ kubectl apply -f kubernetes/deployment.yaml -f kubernetes/rbac.yaml
+$ kubectl apply -f kubernetes/rbac.yaml -f kubernetes/deployment.yaml
 ```
-4. Verify that your ServiceEntries have been populated with the information in Cloud Map; there should be one ServiceEntry for every service in Cloud Map:
+5. Verify that your ServiceEntries have been populated with the information in Cloud Map; there should be one ServiceEntry for every service in Cloud Map:
 ```bash
 $ kubectl get serviceentries
 NAME                                       CREATED AT

cmd/istio-cloud-map/main.go

@@ -101,8 +101,12 @@ func serve() (serve *cobra.Command) {
 			cloudMap := cloudmap.NewStore()
 
 			ctx := context.Background() // common context for cancellation across all loops/routines
-			log.Print("Starting Cloud Map watcher")
-			cmWatcher, err := cloudmap.NewWatcher(cloudMap)
+			awsRegion := os.Getenv("AWS_REGION")
+			if awsRegion == "" {
+				log.Fatal("AWS_REGION env var not set, unable to continue")
+			}
+			log.Printf("Starting Cloud Map watcher in %q", awsRegion)
+			cmWatcher, err := cloudmap.NewWatcher(cloudMap, awsRegion)
 			if err != nil {
 				return err
 			}

kubernetes/deployment.yaml

@@ -27,6 +27,11 @@ spec:
           valueFrom:
             fieldRef:
               fieldPath: metadata.namespace
+        - name: AWS_REGION
+          valueFrom:
+            configMapKeyRef:
+              key: aws-region
+              name: aws-config
         - name: AWS_ACCESS_KEY_ID
           valueFrom:
             secretKeyRef:
@@ -37,3 +42,11 @@ spec:
             secretKeyRef:
               key: secret-access-key
               name: aws-credz
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: aws-config
+  namespace: istio-system
+data:
+  aws-region: us-west-2

pkg/cloudmap/watcher.go

@@ -24,29 +24,19 @@ var serviceFilterNamespaceID = servicediscovery.ServiceFilterNameNamespaceId
 var filterConditionEquals = servicediscovery.FilterConditionEq
 
 // NewWatcher returns a Cloud Map watcher
-func NewWatcher(store Store) (*Watcher, error) {
+func NewWatcher(store Store, region string) (*Watcher, error) {
 	session, err := session.NewSession(&aws.Config{
-		// TODO: env vars aren't a secure way to pass secrets
 		Credentials: credentials.NewEnvCredentials(),
-
-		// TODO: don't hardcode region
-		Region: aws.String("us-west-2"),
+		Region:      aws.String(region),
 	})
 	if err != nil {
 		return nil, errors.Wrap(err, "error setting up AWS session")
-
 	}
-
-	cm := servicediscovery.New(session)
-	cloudmap := servicediscovery.New(session)
-	cloudmap.Endpoint = "https://data-servicediscovery.us-west-2.amazonaws.com"
-
-	return &Watcher{cm: cm, cloudmap: cloudmap, store: store, interval: time.Second * 5}, nil
+	return &Watcher{cloudmap: servicediscovery.New(session), store: store, interval: time.Second * 5}, nil
 }
 
 // Watcher polls Cloud Map and caches a list of services and their instances
 type Watcher struct {
-	cm       servicediscoveryiface.ServiceDiscoveryAPI
 	cloudmap servicediscoveryiface.ServiceDiscoveryAPI
 	store    Store
 	interval time.Duration
@@ -70,7 +60,7 @@ func (w *Watcher) Run(ctx context.Context) {
 func (w *Watcher) refreshStore() {
 	log.Print("Syncing Cloud Map store")
 	// TODO: allow users to specify namespaces to watch
-	nsResp, err := w.cm.ListNamespaces(&servicediscovery.ListNamespacesInput{})
+	nsResp, err := w.cloudmap.ListNamespaces(&servicediscovery.ListNamespacesInput{})
 	if err != nil {
 		log.Printf("error retrieving namespace list from Cloud Map: %v", err)
 		return
@@ -94,7 +84,7 @@ func (w *Watcher) refreshStore() {
 
 func (w *Watcher) hostsForNamespace(ns *servicediscovery.NamespaceSummary) (map[string][]*v1alpha3.ServiceEntry_Endpoint, error) {
 	hosts := map[string][]*v1alpha3.ServiceEntry_Endpoint{}
-	svcResp, err := w.cm.ListServices(&servicediscovery.ListServicesInput{
+	svcResp, err := w.cloudmap.ListServices(&servicediscovery.ListServicesInput{
 		Filters: []*servicediscovery.ServiceFilter{
 			&servicediscovery.ServiceFilter{
 				Name:      &serviceFilterNamespaceID,

pkg/cloudmap/watcher_test.go

@@ -107,7 +107,7 @@ func TestWatcher_refreshCache(t *testing.T) {
 				ListSvcResult: tt.listSvcRes, ListSvcErr: tt.listSvcErr,
 				DiscInstResult: tt.discInstRes, DiscInstErr: tt.discInstErr,
 			}
-			w := &Watcher{cloudmap: mockAPI, cm: mockAPI, store: NewStore()}
+			w := &Watcher{cloudmap: mockAPI, store: NewStore()}
 			w.refreshStore()
 			if !reflect.DeepEqual(w.store.Hosts(), tt.want) {
 				t.Errorf("Watcher.store = %v, want %v", w.store.Hosts(), tt.want)
@@ -163,7 +163,7 @@ func TestWatcher_hostsForNamespace(t *testing.T) {
 				DiscInstResult: tt.discInstRes, DiscInstErr: tt.discInstErr,
 				ListSvcResult: tt.listSvcRes, ListSvcErr: tt.listSvcErr,
 			}
-			w := &Watcher{cloudmap: mockAPI, cm: mockAPI}
+			w := &Watcher{cloudmap: mockAPI}
 			got, err := w.hostsForNamespace(tt.ns)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("Watcher.hostsForNamespace() error = %v, wantErr %v", err, tt.wantErr)

pkg/control/synchronizer.go

@@ -69,7 +69,7 @@ func (s *synchronizer) createOrUpdate(host string, endpoints []*v1alpha3.Service
 			return
 		}
 		// Otherwise, endpoints have changed so update existing Service Entry
-		oldServiceEntry, found := s.istio.Get(model.ServiceEntry.Type, infer.ServiceEntryName(host), "default")
+		oldServiceEntry, found := s.istio.Get(model.ServiceEntry.Type, infer.ServiceEntryName(host), "istio-system")
 		if !found {
 			return
 		}
@@ -96,7 +96,7 @@ func (s *synchronizer) garbageCollect() {
 		if _, ok := s.cloudMap.Hosts()[host]; !ok {
 			// TODO: namespaces!
 			// TODO: Don't attempt to delete no owners
-			if err := s.istio.Delete(model.ServiceEntry.Type, infer.ServiceEntryName(host), "default"); err != nil {
+			if err := s.istio.Delete(model.ServiceEntry.Type, infer.ServiceEntryName(host), "istio-system"); err != nil {
 				log.Printf("error deleting Service Entry %q: %v", infer.ServiceEntryName(host), err)
 			}
 			log.Printf("successfully deleted Service Entry %q", infer.ServiceEntryName(host))

pkg/infer/infer.go

@@ -84,7 +84,6 @@ func Ports(endpoints []*v1alpha3.ServiceEntry_Endpoint) []*v1alpha3.Port {
 
 // Resolution infers STATIC resolution if there are endpoints
 // If there are no endpoints it infers DNS; otherwise will return STATIC
-// TODO: this will probably need to be changed when we support non-IP based addresses
 func Resolution(endpoints []*v1alpha3.ServiceEntry_Endpoint) v1alpha3.ServiceEntry_Resolution {
 	if len(endpoints) == 0 {
 		return v1alpha3.ServiceEntry_DNS