Parity w/0.6.0

- Periodically cleans up old Docker containers
- Periodically reloads master bundle catalog from Cog
- Properly detects when Docker is disabled

New feature: relay.conf is no longer required.

Author: Kevin Smith

main.go

@@ -97,6 +97,8 @@ func tryLoadingConfig(locations []string) config.RawConfig {
 		if err != nil {
 			log.Warnf("Error loading config file '%s': %s.", location, err)
 			continue
+		} else {
+			log.Infof("Using config file '%s'.", location)
 		}
 		return rawConfig
 	}
@@ -146,6 +148,7 @@ func main() {
 		os.Exit(BAD_CONFIG)
 		return
 	}
+	log.Info("%+v", relayConfig)
 	log.Infof("Relay %s is initializing.", relayConfig.ID)
 
 	myRelay, err := relay.NewRelay(relayConfig)

relay/announcer.go

@@ -182,13 +182,11 @@ func (ra *relayAnnouncer) sendAnnouncement(skipTimer bool) {
 
 func getBundles(catalog *bundle.Catalog) []config.Bundle {
 	names := catalog.BundleNames()
-	retval := make([]config.Bundle, len(names))
-	i := 0
+	retval := make([]config.Bundle, 0)
 	for _, name := range names {
 		bundle := catalog.FindLatest(name)
-		if bundle != nil {
-			retval[i] = *bundle
-			i++
+		if bundle != nil && bundle.IsAvailable() {
+			retval = append(retval, *bundle)
 		}
 	}
 	return retval

relay/bus/mqtt.go

@@ -95,7 +95,6 @@ func (mqc *MQTTConnection) buildMQTTOptions(options ConnectionOptions) *mqtt.Cli
 
 func configureSSL(options ConnectionOptions, mqttOpts *mqtt.ClientOptions) error {
 	if !options.SSLEnabled {
-		log.Info("SSL disabled on MQTT connection to Cog")
 		return nil
 	}
 	log.Info("SSL enabled on MQTT connection to Cog")

relay/config/loader.go

@@ -7,9 +7,9 @@ import (
 	"os"
 )
 
-var errorMissingConfigPath = errors.New("Path to configuration file is required.")
-var errorConfigFileNotFound = errors.New("Config file not found or is unreadable.")
-var errorWrongConfigVersion = fmt.Errorf("Only Cog Relay config version %d is supported.", RelayConfigVersion)
+var errorMissingConfigPath = errors.New("Path to configuration file is required")
+var errorConfigFileNotFound = errors.New("Config file not found or is unreadable")
+var errorWrongConfigVersion = fmt.Errorf("Only Cog Relay config version %d is supported", RelayConfigVersion)
 
 // LoadConfig reads a config file off disk
 func LoadConfig(path string) (RawConfig, error) {

relay/engines/docker.go

@@ -30,18 +30,18 @@ type DockerEngine struct {
 
 // NewDockerEngine makes a new DockerEngine instance
 func NewDockerEngine(relayConfig config.Config) (Engine, error) {
-	dockerConfig := relayConfig.Docker
-	if dockerConfig == nil {
+	if relayConfig.DockerEnabled() == false {
 		return nil, errorDockerDisabled
 	}
-	client, err := newClient(*dockerConfig)
+	dockerConfig := *relayConfig.Docker
+	client, err := newClient(dockerConfig)
 	if err != nil {
 		return nil, err
 	}
 	return &DockerEngine{
 		client:      client,
 		relayConfig: relayConfig,
-		config:      *dockerConfig,
+		config:      dockerConfig,
 	}, nil
 }
 

relay/relay.go

@@ -13,6 +13,7 @@ import (
 	"github.com/operable/go-relay/relay/worker"
 	"golang.org/x/net/context"
 	"strings"
+	"time"
 )
 
 const (
@@ -45,6 +46,8 @@ type cogRelay struct {
 	catalog           *bundle.Catalog
 	announcer         Announcer
 	directivesReplyTo string
+	bundleTimer       *time.Timer
+	cleanTimer        *time.Timer
 }
 
 // NewRelay constructs a new Relay instance
@@ -84,10 +87,23 @@ func (r *cogRelay) Start() error {
 	if err := conn.Connect(r.connOpts); err != nil {
 		return err
 	}
+	if r.config.DockerEnabled() {
+		r.cleanTimer = time.AfterFunc(r.config.Docker.CleanDuration(), r.scheduledDockerCleanup)
+		log.Infof("Cleaning up Docker environment on %d second intervals.", r.config.Docker.CleanDuration()/time.Second)
+	}
+	log.Infof("Refreshing bundle catalog on %d second intervals.", r.config.RefreshDuration()/time.Second)
 	return nil
 }
 
 func (r *cogRelay) Stop() error {
+	if r.bundleTimer != nil {
+		r.bundleTimer.Stop()
+	}
+	if r.config.DockerEnabled() {
+		if r.bundleTimer != nil {
+			r.cleanTimer.Stop()
+		}
+	}
 	return nil
 }
 
@@ -134,9 +150,11 @@ func (r *cogRelay) handleCommand(conn bus.Connection, topic string, message []by
 		Payload:     message,
 	}
 	ctx := context.WithValue(context.Background(), "invoke", invoke)
-	log.Debugf("Queue stopped: %v", r.queue.IsStopped())
-	log.Debugf("Eqneueud request: %s", r.queue.Enqueue(ctx))
-	log.Debugf("Enqueued invocation request for %s", topic)
+	if err := r.queue.Enqueue(ctx); err != nil {
+		log.Debugf("Failed enqueuing invocation request: %s.", err)
+	} else {
+		log.Debugf("Enqueued invocation request for %s", topic)
+	}
 }
 
 func (r *cogRelay) handleDirective(conn bus.Connection, topic string, message []byte) {
@@ -148,7 +166,7 @@ func (r *cogRelay) handleDirective(conn bus.Connection, topic string, message []
 	// Dispatch on mesasge type
 	switch tm.(type) {
 	case *messages.ListBundlesResponseEnvelope:
-		log.Info("Processing bundle list")
+		log.Debug("Processing bundle catalog updates.")
 		r.updateCatalog(tm.(*messages.ListBundlesResponseEnvelope))
 	}
 }
@@ -174,21 +192,34 @@ func (r *cogRelay) updateCatalog(envelope *messages.ListBundlesResponseEnvelope)
 	// TODO: This should be bi-directional sync to catch bundle unassignments too
 	r.catalog.AddBatch(bundles)
 	if r.catalog.IsChanged() {
-		r.refreshBundles()
-		log.Info("Changes to bundle assignments detected.")
-		r.announcer.SendAnnouncement()
+		if err := r.refreshBundles(); err != nil {
+			log.Errorf("Bundle catalog refresh failed: %s.", err)
+		} else {
+			log.Info("Changes to bundle catalog detected.")
+			r.announcer.SendAnnouncement()
+		}
 	}
+	r.bundleTimer = time.AfterFunc(r.config.RefreshDuration(), r.scheduledBundleRefresh)
 }
 
 func (r *cogRelay) refreshBundles() error {
 	dockerEngine, err := engines.NewDockerEngine(*r.config)
 	if err != nil {
-		return err
+		if r.config.DockerEnabled() == false {
+			dockerEngine = nil
+		} else {
+			return err
+		}
 	}
 	for _, name := range r.catalog.BundleNames() {
 		if bundle := r.catalog.FindLatest(name); bundle != nil {
 			if bundle.NeedsRefresh() {
 				if bundle.IsDocker() {
+					if r.config.DockerEnabled() == false {
+						log.Infof("Skipping Docker-based bundle %s %s.", bundle.Name, bundle.Version)
+						bundle.SetAvailable(false)
+						continue
+					}
 					avail, _ := dockerEngine.IsAvailable(bundle.Docker.Image, bundle.Docker.Tag)
 					bundle.SetAvailable(avail)
 				} else {
@@ -208,10 +239,34 @@ func (r *cogRelay) requestBundles() error {
 		},
 	}
 	raw, _ := json.Marshal(&msg)
-	log.Info("Refreshing command bundles.")
+	log.Debug("Refreshing command catalog.")
 	return r.conn.Publish(infoTopic, raw)
 }
 
+func (r *cogRelay) scheduledBundleRefresh() {
+	if err := r.requestBundles(); err != nil {
+		log.Errorf("Scheduled bundle catalog refresh failed: %s.", err)
+		r.bundleTimer = time.AfterFunc(r.config.RefreshDuration(), r.scheduledBundleRefresh)
+	}
+}
+
+func (r *cogRelay) scheduledDockerCleanup() {
+	engine, err := engines.NewDockerEngine(*r.config)
+	if err != nil {
+		log.Errorf("Scheduled clean up of Docker environment failed: %s.", err)
+	} else {
+		cleaned := engine.Clean()
+		container := "containers"
+		if cleaned == 1 {
+			container = "container"
+		}
+		if cleaned > 0 {
+			log.Infof("Scheduled Docker clean up removed %d %s.", cleaned, container)
+		}
+	}
+	r.cleanTimer = time.AfterFunc(r.config.Docker.CleanDuration(), r.scheduledDockerCleanup)
+}
+
 func verifyDockerConfig(c *config.Config) error {
 	if c.DockerEnabled() == true {
 		if err := engines.VerifyDockerConfig(c.Docker); err != nil {