clean up aws plugin; fix up environment variables

Signed-off-by: David Chung <[email protected]>

Author: David Chung

pkg/provider/aws/plugin/instance/monitor.go

@@ -18,8 +18,9 @@ const (
 // list of known instances, and report anything it hasn't seen before, or
 // if anything that disappeared.
 type Monitor struct {
-	stop   chan struct{}
-	topics map[string]interface{}
+	PollInterval time.Duration
+	stop         chan struct{}
+	topics       map[string]interface{}
 
 	// Plugin is the instance plugin to use
 	Plugin instance.Plugin
@@ -64,7 +65,10 @@ func (m *Monitor) PublishOn(c chan<- *event.Event) {
 
 		log.Infoln("Start monitoring instances", c)
 
-		ticker := time.Tick(2 * time.Second)
+		ticker := time.Tick(5 * time.Second)
+		if m.PollInterval > 0 {
+			ticker = time.Tick(m.PollInterval)
+		}
 
 		instances := map[instance.ID]instance.Description{}
 		last := mapset.NewSet()

pkg/rpc/client/client.go

@@ -8,13 +8,14 @@ import (
 	"net/http"
 	"net/http/httputil"
 	"net/url"
-	"os"
+	//	"os"
 	"path"
 	"sync"
 	"time"
 
 	logutil "github.com/docker/infrakit/pkg/log"
 	"github.com/docker/infrakit/pkg/rpc"
+	"github.com/docker/infrakit/pkg/run/local"
 	"github.com/docker/infrakit/pkg/spi"
 	"github.com/gorilla/rpc/v2/json2"
 )
@@ -74,17 +75,6 @@ func New(address string, api spi.InterfaceSpec) (Client, error) {
 	return cl, nil
 }
 
-// ClientTimeoutEnv environment variable for the client timeout (in time.Duration)
-const ClientTimeoutEnv = "INFRAKIT_CLIENT_TIMEOUT"
-
-// timeout returns the client rpc http timeout
-func timeout() time.Duration {
-	if parsed, err := time.ParseDuration(os.Getenv(ClientTimeoutEnv)); err == nil {
-		return parsed
-	}
-	return time.Duration(1 * time.Second)
-}
-
 func parseAddress(address string) (*url.URL, *http.Client, error) {
 	if path.Ext(address) == ".listen" {
 		buff, err := ioutil.ReadFile(address)
@@ -105,7 +95,7 @@ func parseAddress(address string) (*url.URL, *http.Client, error) {
 		u.Host = "h"
 		u.Path = "" // clear it since it's a file path and we are using it to connect.
 		return u, &http.Client{
-			Timeout: timeout(),
+			Timeout: local.ClientTimeout(),
 			Transport: &http.Transport{
 				// TODO(chungers) - fix this deprecation
 				Dial: func(proto, addr string) (conn net.Conn, err error) {
@@ -122,9 +112,9 @@ func parseAddress(address string) (*url.URL, *http.Client, error) {
 	case "http", "https":
 		transport := &http.Transport{
 			Dial: (&net.Dialer{
-				Timeout: timeout(),
+				Timeout: local.ClientTimeout(),
 			}).Dial,
-			TLSHandshakeTimeout: timeout(),
+			TLSHandshakeTimeout: local.ClientTimeout(),
 		}
 		return u, &http.Client{Transport: transport}, nil
 

pkg/run/local/local.go

@@ -5,6 +5,9 @@ import (
 	"os"
 	"os/user"
 	"path/filepath"
+	"time"
+
+	"github.com/docker/infrakit/pkg/types"
 )
 
 const (
@@ -14,8 +17,16 @@ const (
 
 	// EnvPlaybooks is the environment variable for storing the playbooks file
 	EnvPlaybooks = "INFRAKIT_PLAYBOOKS_FILE"
+
+	// EnvClientTimeout is the timeout used by the rpc client
+	EnvClientTimeout = "INFRAKIT_CLIENT_TIMEOUT"
 )
 
+// ClientTimeout returns the client timeout
+func ClientTimeout() time.Duration {
+	return types.MustParseDuration(Getenv(EnvClientTimeout, "15s")).Duration()
+}
+
 // InfrakitHome returns the directory of INFRAKIT_HOME if specified. Otherwise, it will return
 // the user's home directory.  If that cannot be determined, then it returns the current working
 // directory.  If that still cannot be determined, a temporary directory is returned.

pkg/run/v0/aws/aws.go

@@ -2,6 +2,7 @@ package aws
 
 import (
 	"strings"
+	"time"
 
 	"github.com/aws/aws-sdk-go/service/autoscaling"
 	"github.com/aws/aws-sdk-go/service/cloudwatchlogs"
@@ -34,14 +35,17 @@ const (
 	EnvRegion = "INFRAKIT_AWS_REGION"
 
 	// EnvStackName is the env for stack name
-	EnvStackName = "INFRAKIT_AWS_STACKNAME"
+	EnvStackName = "INFRAKIT_AWS_STACK_NAME"
 
 	// EnvMetadataTemplateURL is the location of the template for Metadata plugin
 	EnvMetadataTemplateURL = "INFRAKIT_AWS_METADATA_TEMPLATE_URL"
 
 	// EnvMetadataPollInterval is the env to set fo polling for metadata updates
 	EnvMetadataPollInterval = "INFRAKIT_AWS_METADATA_POLL_INTERVAL"
 
+	// EnvMonitorPollInterval is the env to set fo polling for instance changes
+	EnvMonitorPollInterval = "INFRAKIT_AWS_MONITOR_POLL_INTERVAL"
+
 	// EnvNamespaceTags is the env to set for namespace tags. It's k=v,...
 	EnvNamespaceTags = "INFRAKIT_AWS_NAMESPACE_TAGS"
 
@@ -59,12 +63,16 @@ func init() {
 
 // Options capture the options for starting up the plugin.
 type Options struct {
+
 	// Namespace is a set of kv pairs for tags that namespaces the resource instances
 	Namespace map[string]string
 
 	// ELBNames is a list of names for ELB instances to start the L4 plugins
 	ELBNames []string
 
+	// MonitorPollInterval is the interval for the polling to observe instance new/delete events
+	MonitorPollInterval time.Duration
+
 	aws_metadata.Options `json:",inline" yaml:",inline"`
 }
 
@@ -82,13 +90,16 @@ func defaultNamespace() map[string]string {
 
 // DefaultOptions return an Options with default values filled in.
 var DefaultOptions = Options{
-	Namespace: defaultNamespace(),
-	ELBNames:  strings.Split(local.Getenv(EnvELBNames, ""), ","),
+	Namespace:           defaultNamespace(),
+	ELBNames:            strings.Split(local.Getenv(EnvELBNames, ""), ","),
+	MonitorPollInterval: types.MustParseDuration(local.Getenv(EnvMonitorPollInterval, "0s")).Duration(),
 	Options: aws_metadata.Options{
 		Template:  local.Getenv(EnvMetadataTemplateURL, ""),
 		StackName: local.Getenv(EnvStackName, ""),
 		Options: aws_instance.Options{
-			Region: local.Getenv(EnvRegion, ""), // empty string trigger auto-detect
+			Region:          local.Getenv(EnvRegion, ""), // empty string trigger auto-detect
+			AccessKeyID:     local.Getenv("AWS_ACCESS_KEY_ID", ""),
+			SecretAccessKey: local.Getenv("AWS_SECRET_ACCESS_KEY", ""),
 		},
 		PollInterval: types.MustParseDuration(local.Getenv(EnvMetadataPollInterval, "60s")),
 	},
@@ -105,27 +116,13 @@ func Run(scope scope.Scope, name plugin.Name,
 		return
 	}
 
-	var metadataPlugin metadata.Plugin
-	stopMetadataPlugin := make(chan struct{})
-	metadataPlugin, err = aws_metadata.NewPlugin(options.Options, stopMetadataPlugin)
-	if err != nil {
-		return
-	}
-
 	var instancePlugin instance.Plugin
 	builder := aws_instance.Builder{Options: options.Options.Options}
 	instancePlugin, err = builder.BuildInstancePlugin(options.Namespace)
 	if err != nil {
 		return
 	}
 
-	monitor := &aws_instance.Monitor{Plugin: instancePlugin}
-
-	onStop = func() {
-		close(stopMetadataPlugin)
-		monitor.Stop()
-	}
-
 	autoscalingClient := autoscaling.New(builder.Config)
 	cloudWatchLogsClient := cloudwatchlogs.New(builder.Config)
 	dynamodbClient := dynamodb.New(builder.Config)
@@ -136,10 +133,6 @@ func Run(scope scope.Scope, name plugin.Name,
 
 	transport.Name = name
 	impls = map[run.PluginCode]interface{}{
-		run.Event: map[string]event.Plugin{
-			"ec2-instance": monitor.Init(),
-		},
-		run.Metadata: metadataPlugin,
 		run.Instance: map[string]instance.Plugin{
 			"autoscaling-autoscalinggroup":    aws_instance.NewAutoScalingGroupPlugin(autoscalingClient, options.Namespace),
 			"autoscaling-launchconfiguration": aws_instance.NewLaunchConfigurationPlugin(autoscalingClient, options.Namespace),
@@ -175,5 +168,37 @@ func Run(scope scope.Scope, name plugin.Name,
 		impls[run.L4] = func() (map[string]loadbalancer.L4, error) { return l4Map, nil }
 	}
 
+	if options.MonitorPollInterval > 0 {
+		log.Info("run the event source for watching instance add/removes", "poll", options.MonitorPollInterval)
+		monitor := &aws_instance.Monitor{Plugin: instancePlugin}
+		impls[run.Event] = map[string]event.Plugin{
+			"ec2-instance": monitor.Init(),
+		}
+		onStop = func() {
+			monitor.Stop()
+		}
+	}
+
+	if u := local.Getenv(EnvMetadataTemplateURL, ""); u != "" {
+
+		log.Info("Include metadata plugin", "url", u)
+
+		var metadataPlugin metadata.Plugin
+		stopMetadataPlugin := make(chan struct{})
+		metadataPlugin, err = aws_metadata.NewPlugin(options.Options, stopMetadataPlugin)
+		if err != nil {
+			return
+		}
+		impls[run.Metadata] = metadataPlugin
+
+		cleanup := onStop
+		onStop = func() {
+			close(stopMetadataPlugin)
+			if cleanup != nil {
+				cleanup()
+			}
+		}
+	}
+
 	return
 }